Skip to content


added basic module autodetection
Browse files Browse the repository at this point in the history
  • Loading branch information
ErikBjare committed Aug 30, 2017
1 parent 52b2837 commit 53f7dd0
Showing 1 changed file with 91 additions and 37 deletions.
128 changes: 91 additions & 37 deletions aw_qt/
Original file line number Diff line number Diff line change
@@ -1,37 +1,80 @@
import os
import platform
from glob import glob
from time import sleep
import logging
import subprocess
import shutil
from typing import Optional, List

import aw_core

logger = logging.getLogger(__name__)

_module_dir = os.path.dirname(os.path.realpath(__file__))
_parent_dir = os.path.abspath(os.path.join(_module_dir, os.pardir))
_search_paths = [_module_dir, _parent_dir]

def _locate_executable(name: str) -> List[str]:
Will start module from localdir if present there,
otherwise will try to call what is available in PATH.

Returns it as a Popen cmd list.
curr_filepath = os.path.realpath(__file__)
curr_dir = os.path.dirname(curr_filepath)
search_paths = [curr_dir, os.path.abspath(os.path.join(curr_dir, os.pardir))]
exec_paths = [os.path.join(path, name) for path in search_paths]
def _locate_bundled_executable(name: str) -> Optional[str]:
"""Returns the path to the module executable if it exists in the bundle, else None."""
_exec_paths = [os.path.join(path, name) for path in _search_paths]

for exec_path in exec_paths:
# Look for it in the installation path
for exec_path in _exec_paths:
if os.path.isfile(exec_path):
# logger.debug("Found executable for {} in: {}".format(name, exec_path))
return [exec_path]
break # this break is redundant, but kept due to for-else semantics
return exec_path

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

def _locate_executable(name: str) -> Optional[str]:
Will return the path to the executable if bundled,
otherwise returns the name if it is available in PATH.
Used when calling Popen.
exec_path = _locate_bundled_executable(name)
if exec_path is not None: # Check if it exists in bundle
return exec_path
elif _is_system_module(name): # Check if it's in PATH
return name
# TODO: Actually check if it is in PATH
# logger.debug("Trying to start {} using PATH (executable not found in: {})"
# .format(name, exec_paths))
return [name]
logger.warning("Could not find module '{}' in installation directory or PATH".format(name))
return None

def _discover_modules_bundled() -> List[str]:
# Look for modules in source dir and parent dir
modules = []
for path in _search_paths:
matches = glob(os.path.join(path, "aw-*"))
for match in matches:
if os.path.isfile(match) and os.access(match, os.X_OK):
logger.warning("Found matching file but was not executable: {}".format(path))"Found bundled modules: {}".format(set(modules)))
return modules

def _discover_modules_system() -> List[str]:
search_paths = os.environ["PATH"].split(":")
modules = []
for path in search_paths:
files = os.listdir(path)
for filename in files:
if "aw-" in filename:
modules.append(filename)"Found system modules: {}".format(set(modules)))
return modules

class Module:
Expand All @@ -46,20 +89,25 @@ def start(self) -> None:"Starting module {}".format(

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

exec_cmd = _locate_executable(
if self.testing:
# logger.debug("Running: {}".format(exec_cmd))
exec_path = _locate_executable(
if exec_path is None:
exec_cmd = [exec_path]
if self.testing:
# logger.debug("Running: {}".format(exec_cmd))

# There is a very good reason stdout and stderr is not PIPE here
# See:
self._process = subprocess.Popen(exec_cmd, universal_newlines=True)
# There is a very good reason stdout and stderr is not PIPE here
# See:
self._process = subprocess.Popen(exec_cmd, universal_newlines=True)

# Should be True if module is supposed to be running, else False
self.started = True
# Should be True if module is supposed to be running, else False
self.started = True

def stop(self) -> None:
Expand Down Expand Up @@ -108,19 +156,25 @@ def read_log(self) -> str:

class Manager:
def __init__(self, testing: bool=False):
# TODO: Fetch these from somewhere appropriate (auto detect or a config file)
# Save to config wether they should autostart or not.
_possible_modules = [
def __init__(self, testing: bool = False) -> None:
self.testing = testing
self.modules = {} # type: Dict[str, Module]


def discover_modules(self):
# These should always be bundled with aw-qt
found_modules = {
# "aw-watcher-spotify",
# "aw-watcher-network"

# TODO: Filter away all modules not available on system
self.modules = {name: Module(name, testing=testing) for name in _possible_modules}
found_modules |= set(_discover_modules_bundled())
found_modules |= set(_discover_modules_system())

for m_name in found_modules:
if m_name not in self.modules:
self.modules[m_name] = Module(m_name, testing=self.testing)

def get_unexpected_stops(self):
return list(filter(lambda x: x.started and not x.is_alive(), self.modules.values()))
Expand Down

0 comments on commit 53f7dd0

Please sign in to comment.