Archives/input-remapper
2
0
Fork 0
mirror of https://github.com/sezanzeb/input-remapper synced 2024-11-10 01:13:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

#197 Not attempting to autoload if no user is logged in yet

Browse Source
This commit is contained in:
sezanzeb 2021-11-22 23:19:51 +01:00
parent 3fb42ad438
commit ef3970f33e
5 changed files with 56 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

71
bin/key-mapper-control
View File

@ -27,6 +27,7 @@ import grp
import sys
import argparse
import logging
import psutil
from keymapper.logger import logger, update_verbosity, log_info, add_filehandler
from keymapper.config import config
@ -99,7 +100,10 @@ def communicate(options, daemon):
group = groups.find(key=options.device)
if group is None:
logger.error('Unknown device "%s"', options.device)
logger.error(
'Device "%s" is unknown or not an appropriate input device',
options.device
)
sys.exit(1)
return group
@ -139,7 +143,7 @@ def communicate(options, daemon):
daemon.autoload(timeout=10)
else:
group = require_group()
logger.info('Autoloading %s', options.device)
logger.info('Asking daemon to autoload for %s', options.device)
daemon.autoload_single(group.key, timeout=2)
if options.command == START:
@ -183,6 +187,43 @@ def internals(options):
os.system(cmd)
def main(options):
if options.debug:
update_verbosity(True)
add_filehandler('/var/log/key-mapper-control')
if options.version:
log_info()
return
logger.debug('Call for "%s"', sys.argv)
if options.command == AUTOLOAD and len(psutil.users()) == 0:
# No user logged in, this is probably happening during boot time and got
# triggered by udev. There is no need to try to inject anything if the
# service doesn't know where to look for a config file. This avoids a lot
# of confusing service logs. And also avoids potential for problems when
# key-mapper-control stresses about evdev, dbus and multiprocessing already
# while the system hasn't even booted completely.
logger.warn('Skipping autoload command without a logged in user')
return
if options.command is not None:
if options.command in INTERNALS:
internals(options)
elif options.command in COMMANDS:
from keymapper.daemon import Daemon
daemon = Daemon.connect(fallback=False)
communicate(options, daemon)
else:
logger.error('Unknown command "%s"', options.command)
else:
utils(options)
if options.command:
logger.info('Finished')
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument(
@ -230,28 +271,4 @@ if __name__ == '__main__':
help='Print the version and exit', default=False
)
options = parser.parse_args(sys.argv[1:])
if options.debug:
update_verbosity(True)
add_filehandler('/var/log/key-mapper-control')
if options.version:
log_info()
else:
logger.debug('Call for "%s"', sys.argv)
if options.command is not None:
if options.command in INTERNALS:
internals(options)
elif options.command in COMMANDS:
from keymapper.daemon import Daemon
daemon = Daemon.connect(fallback=False)
communicate(options, daemon)
else:
logger.error('Unknown command "%s"', options.command)
else:
utils(options)
if options.command:
logger.info('Finished')
main(parser.parse_args(sys.argv[1:]))

6
keymapper/groups.py
View File

@ -34,7 +34,6 @@ and the injector.
import re
import multiprocessing
import threading
import time
import asyncio
import json
@ -484,7 +483,7 @@ class _Groups:
"""Load a serialized representation created via dumps."""
self._groups = [_Group.loads(group) for group in json.loads(dump)]
def find(self, name=None, key=None, path=None):
def find(self, name=None, key=None, path=None, include_keymapper=False):
"""Find a group that matches the provided parameters.
Parameters
@ -498,6 +497,9 @@ class _Groups:
"/dev/input/event3"
"""
for group in self._groups:
if not include_keymapper and group.name.startswith("key-mapper"):
continue
if name and group.name != name:
continue

2
keymapper/injection/consumer_control.py
View File

@ -40,7 +40,7 @@ consumer_classes = [
class ConsumerControl:
"""Reads input events from a single device and distributes them.
There is one Events object for each source, which tells multiple consumers
There is one ConsumerControl object for each source, which tells multiple consumers
that a new event is ready so that they can inject all sorts of funny
things.

7
keymapper/logger.py
View File

@ -214,7 +214,7 @@ def add_filehandler(log_path=LOG_PATH):
if os.path.exists(log_path):
# the logfile should not be too long to avoid overflowing the storage
with open(log_path, "r") as file:
content = file.readlines()[-1000:] + ["---\n"]
content = file.readlines()[-1000:] + ["\n"]
with open(log_path, "w") as file:
file.truncate(0)
@ -222,9 +222,8 @@ def add_filehandler(log_path=LOG_PATH):
file_handler = logging.FileHandler(log_path)
file_handler.setFormatter(Formatter())
logger.info('Logging to "%s"', log_path)
logger.addHandler(file_handler)
logger.info('Starting logging to "%s" at %s', log_path, str(datetime.now()))
except PermissionError:
logger.debug('No permission to log to "%s"', log_path)

8
tests/testcases/test_logger.py
View File

@ -77,15 +77,15 @@ class TestLogger(unittest.TestCase):
os.mknod(path)
with open(path, "w") as f:
f.write("line\n" * 2000 + "end")
f.write("aaaa\n" * 2000 + "end")
add_filehandler(os.path.join(tmp, "logger-test"))
with open(path, "r") as f:
# it only keeps the newest information
content = f.readlines()
self.assertLess(len(content), 1100)
self.assertEqual(content[-1], "end---\n")
# after "---" new log will appear
# whatever the logging module decides to log into that file
self.assertNotIn("aaaa", content[-1])
def test_debug(self):
path = os.path.join(tmp, "logger-test")
@ -116,8 +116,8 @@ class TestLogger(unittest.TestCase):
def test_default(self):
path = os.path.join(tmp, "logger-test")
add_filehandler(path)
update_verbosity(debug=False)
add_filehandler(path)
logger.error("abc")
logger.warning("foo")
logger.info("123")