Skip to content


Improve mypy typing
Browse files Browse the repository at this point in the history
  • Loading branch information
xylix committed Mar 4, 2020
1 parent 2a05ba1 commit 4403d19
Show file tree
Hide file tree
Showing 5 changed files with 63 additions and 47 deletions.
2 changes: 1 addition & 1 deletion Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ test-integration:
python3 ./tests/ --no-modules

mypy aw_qt --ignore-missing-imports
mypy aw_qt --strict

make typecheck
Expand Down
23 changes: 14 additions & 9 deletions aw_qt/
Original file line number Diff line number Diff line change
@@ -1,36 +1,41 @@
import sys
import logging
import argparse
from typing import Set
from typing_extensions import TypedDict

from aw_core.log import setup_logging

from .manager import Manager

from . import trayicon

logger = logging.getLogger(__name__)

def main():
def main() -> None:
args = parse_args()
setup_logging("aw-qt", testing=args.testing, verbose=args.testing, log_file=True)
setup_logging("aw-qt", testing=args['testing'], verbose=args['testing'], log_file=True)

_manager = Manager(testing=args.testing)
_manager = Manager(testing=args['testing'])

error_code =, testing=args.testing)
error_code =, testing=args['testing'])


def parse_args():
CommandLineArgs = TypedDict('CommandLineArgs', {'testing': bool, 'autostart_modules': Set[str]}, total=False)

def parse_args() -> CommandLineArgs:
parser = argparse.ArgumentParser(prog="aw-qt", description='A trayicon and service manager for ActivityWatch')
parser.add_argument('--testing', action='store_true',
help='Run the trayicon and services in testing mode')
parser.add_argument('--autostart-modules', dest='autostart_modules',
type=lambda s: [m for m in s.split(',') if m and m.lower() != "none"],
help='A comma-separated list of modules to autostart, or just `none` to not autostart anything')

return parser.parse_args()
parsed_args = parser.parse_args()
dict: CommandLineArgs = {'autostart_modules': parsed_args.autostart_modules, 'testing': parsed_args.testing}
return dict
34 changes: 18 additions & 16 deletions aw_qt/
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ def _locate_bundled_executable(name: str) -> Optional[str]:
return None

def _is_system_module(name) -> bool:
def _is_system_module(name: str) -> bool:
"""Checks if a module with a particular name exists in PATH"""
return shutil.which(name) is not None

Expand Down Expand Up @@ -88,20 +88,20 @@ def __init__(self, name: str, testing: bool = False) -> None:
self.started = False # Should be True if module is supposed to be running, else False
self.testing = testing
self.location = "system" if _is_system_module(name) else "bundled"
self._process = None # type: Optional[subprocess.Popen]
self._last_process = None # type: Optional[subprocess.Popen]
self._process: Optional[subprocess.Popen[str]] = None
self._last_process: Optional[subprocess.Popen[str]] = None

def start(self) -> None:"Starting module {}".format(

# Create a process group, become its leader
# TODO: This shouldn't go here
if platform.system() != "Windows":
os.setpgrp() # type: ignore

exec_path = _locate_executable(
if exec_path is None:
logger.error("Tried to start nonexistent module {}".format(
exec_cmd = [exec_path]
if self.testing:
Expand All @@ -112,8 +112,8 @@ def start(self) -> None:
# See:
startupinfo = None
if platform.system() == "Windows":
startupinfo = subprocess.STARTUPINFO() # type: ignore
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW # type: ignore
startupinfo = subprocess.STARTUPINFO() #type: ignore
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW #type: ignore
elif platform.system() == "Darwin":"Macos: Disable dock icon")
import AppKit
Expand All @@ -122,7 +122,6 @@ def start(self) -> None:
# There is a very good reason stdout and stderr is not PIPE here
# See:
self._process = subprocess.Popen(exec_cmd, universal_newlines=True, startupinfo=startupinfo)

self.started = True

def stop(self) -> None:
Expand All @@ -137,7 +136,7 @@ def stop(self) -> None:
logger.warning("Tried to stop module {}, but it wasn't running".format(
if not self._process:
logger.error("No reference to process object")
logger.debug("Stopping module {}".format(
if self._process:
Expand Down Expand Up @@ -180,6 +179,7 @@ def __init__(self, testing: bool = False) -> None:
self.settings: AwQtSettings = AwQtSettings(testing)
self.modules: Dict[str, Module] = {}
self.autostart_modules: Set[str] = set(self.settings.autostart_modules)
self.testing = testing

for name in self.settings.possible_modules:
if _locate_executable(name):
Expand All @@ -189,7 +189,7 @@ def __init__(self, testing: bool = False) -> None:
# Is this actually a good way to do this? merged from dev/autodetect-modules

def discover_modules(self):
def discover_modules(self) -> None:
# These should always be bundled with aw-qt
found_modules = set(_discover_modules_bundled())
found_modules |= set(_discover_modules_system())
Expand All @@ -199,17 +199,19 @@ def discover_modules(self):
if m_name not in self.modules:
self.modules[m_name] = Module(m_name, testing=self.testing)

def get_unexpected_stops(self):
def get_unexpected_stops(self) -> List[Module]:
return list(filter(lambda x: x.started and not x.is_alive(), self.modules.values()))

def start(self, module_name):
def start(self, module_name: str) -> None:
if module_name in self.modules.keys():
logger.debug("Manager tried to start nonexistent module {}".format(module_name))

def autostart(self, autostart_modules):
if autostart_modules is None:
def autostart(self, autostart_modules: Set[str] = set()) -> None:
if autostart_modules and len(autostart_modules) > 0:"Modules to start weren't specified in CLI arguments. Falling back to configuration.")
autostart_modules = self.settings.autostart_modules
autostart_modules = set(self.settings.autostart_modules)

# We only want to autostart modules that are both in found modules and are asked to autostart.
autostart_modules = autostart_modules.intersection(set(self.modules.keys()))
Expand All @@ -223,7 +225,7 @@ def autostart(self, autostart_modules):
for module_name in autostart_modules:

def stop_all(self):
def stop_all(self) -> None:
for module in filter(lambda m: m.is_alive(), self.modules.values()):

Expand Down
45 changes: 24 additions & 21 deletions aw_qt/
Original file line number Diff line number Diff line change
Expand Up @@ -5,29 +5,30 @@
import os
import subprocess
from collections import defaultdict
from typing import Any, DefaultDict, List, Optional

from PyQt5 import QtCore
from PyQt5.QtWidgets import QApplication, QSystemTrayIcon, QMessageBox, QMenu, QWidget, QPushButton
from PyQt5.QtGui import QIcon

import aw_core

from .manager import Manager
from .manager import Manager, Module

logger = logging.getLogger(__name__)

def open_webui(root_url):
def open_webui(root_url: str) -> None:
print("Opening dashboard")

def open_apibrowser(root_url):
def open_apibrowser(root_url: str) -> None:
print("Opening api browser") + "/api")

def open_dir(d):
def open_dir(d: str)-> None:
if sys.platform == 'win32':
Expand All @@ -38,17 +39,18 @@ def open_dir(d):

class TrayIcon(QSystemTrayIcon):
def __init__(self, manager: Manager, icon, parent=None, testing=False) -> None:
def __init__(self, manager: Manager, icon: QIcon, parent: Optional[QWidget]=None, testing: bool=False) -> None:
QSystemTrayIcon.__init__(self, icon, parent)
self._parent = parent # QSystemTrayIcon also tries to save parent info but it screws up the type info
self.setToolTip("ActivityWatch" + (" (testing)" if testing else ""))

self.manager = manager
self.testing = testing


def _build_rootmenu(self):
menu = QMenu(self.parent())
def _build_rootmenu(self) -> None:
menu = QMenu(self._parent)

root_url = "http://localhost:{port}".format(port=5666 if self.testing else 5600)

Expand Down Expand Up @@ -80,8 +82,8 @@ def _build_rootmenu(self):


def show_module_failed_dialog(module):
box = QMessageBox(self.parent())
def show_module_failed_dialog(module: Module) -> None:
box = QMessageBox(self._parent)
box.setText("Module {} quit unexpectedly".format(
Expand All @@ -93,10 +95,10 @@ def show_module_failed_dialog(module):

def rebuild_modules_menu():
def rebuild_modules_menu() -> None:
for action in modulesMenu.actions():
if action.isEnabled():
name =
name =
alive = self.manager.modules[name].is_alive()
# print(module.text(), alive)
Expand All @@ -105,7 +107,7 @@ def rebuild_modules_menu():
QtCore.QTimer.singleShot(2000, rebuild_modules_menu)
QtCore.QTimer.singleShot(2000, rebuild_modules_menu)

def check_module_status():
def check_module_status() -> None:
unexpected_exits = self.manager.get_unexpected_stops()
if unexpected_exits:
for module in unexpected_exits:
Expand All @@ -116,19 +118,19 @@ def check_module_status():
QtCore.QTimer.singleShot(2000, rebuild_modules_menu)
QtCore.QTimer.singleShot(2000, check_module_status)

def _build_modulemenu(self, moduleMenu):
def _build_modulemenu(self, moduleMenu: QMenu) -> None:

def add_module_menuitem(module):
def add_module_menuitem(module: Module) -> None:
title =
ac = moduleMenu.addAction(title, lambda: module.toggle())
# Kind of nasty, but a quick way to affiliate the module to the menu action for when it needs updating
ac.module = module


# Merged from branch dev/autodetect-modules, still kind of in progress with making this actually work
modules_by_location = defaultdict(lambda: list())
modules_by_location: DefaultDict[str, List[Module]] = defaultdict(lambda: list())
for module in sorted(self.manager.modules.values(), key=lambda m:

Expand All @@ -140,7 +142,7 @@ def add_module_menuitem(module):

def exit(manager: Manager):
def exit(manager: Manager) -> None:
# TODO: Do cleanup actions
# TODO: Save state for resume
print("Shutdown initiated, stopping all services...")
Expand All @@ -151,16 +153,17 @@ def exit(manager: Manager):

def run(manager, testing=False):
def run(manager: Manager, testing: bool = False) -> Any:"Creating trayicon...")
# print(QIcon.themeSearchPaths())

app = QApplication(sys.argv)

# FIXME: remove ignores after has been fixed
# Without this, Ctrl+C will have no effect
signal.signal(signal.SIGINT, lambda: exit(manager))
signal.signal(signal.SIGINT, lambda: exit(manager)) #type: ignore
# Ensure cleanup happens on SIGTERM
signal.signal(signal.SIGTERM, lambda: exit(manager))
signal.signal(signal.SIGTERM, lambda: exit(manager)) #type: ignore

timer = QtCore.QTimer()
timer.start(100) # You may change this if you wish.
Expand Down
6 changes: 6 additions & 0 deletions mypy.ini
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
python_version = 3.6
ignore_missing_imports = True

ignore_errors = True

0 comments on commit 4403d19

Please sign in to comment.