From 6d0ede9ea0b375aba7f83d099857c475554f6ecd Mon Sep 17 00:00:00 2001 From: Daniel Schep Date: Tue, 24 Jan 2017 22:58:02 -0500 Subject: [PATCH] WIP of #125: screen locking detection separate backends implemented via a 'multi' meta-backend. Uses config like this: ``` backends: [multi] multi: locked: pushover: user_key: user-api-key unfocused: default: {} focused: {} ``` This config would cause no notifications if the shell is focused, desktop notifications when unfocused, and pushover notifications when the screen is locked. Freaking magic. :tada: --- ntfy/backends/multi.py | 25 ++++++++++ ntfy/cli.py | 15 ++++++ ntfy/screensaver.py | 106 +++++++++++++++++++++++++++++++++++++++++ tests/test_cli.py | 11 +++++ 4 files changed, 157 insertions(+) create mode 100644 ntfy/backends/multi.py create mode 100644 ntfy/screensaver.py diff --git a/ntfy/backends/multi.py b/ntfy/backends/multi.py new file mode 100644 index 0000000..046bc86 --- /dev/null +++ b/ntfy/backends/multi.py @@ -0,0 +1,25 @@ +from importlib import import_module +try: + from ..terminal import is_focused +except ImportError: + def is_focused(): + return True +from ..screensaver import is_locked + + +def notify(title, + message, + locked=None, + focused=None, + unfocused=None, + retcode=None): + for condition, options in ((is_locked, locked), + (is_focused, focused), + (lambda: not is_focused(), unfocused)): + for backend_name, backend_options in options.items(): + if not condition(): + continue + backend = import_module('ntfy.backends.{}'.format( + backend_options.get('backend', backend_name))) + backend_options.pop('backend', None) + backend.notify(title, message, retcode=retcode, **backend_options) diff --git a/ntfy/cli.py b/ntfy/cli.py index bd88e2a..d149df5 100644 --- a/ntfy/cli.py +++ b/ntfy/cli.py @@ -31,6 +31,7 @@ def is_focused(): return True +from .screensaver import is_locked def run_cmd(args): @@ -62,6 +63,8 @@ def run_cmd(args): retcode = process.returncode if args.longer_than is not None and duration <= args.longer_than: return None, None + if args.locked_only and not is_locked(): + return None, None if args.unfocused_only and is_focused(): return None, None message = _result_message(args.command if not args.hide_command else None, @@ -228,6 +231,12 @@ def default_sender(args): type=int, metavar='N', help="Only notify if the command runs longer than N seconds") +done_parser.add_argument( + '--locked-only', + action='store_true', + default=False, + dest='locked_only', + help='Only notify if the screen is locked') done_parser.add_argument( '-b', '--background-only', @@ -281,6 +290,12 @@ def default_sender(args): type=int, metavar='N', help="Only notify if the command runs longer than N seconds") +shell_integration_parser.add_argument( + '--locked-only', + action='store_true', + default=False, + dest='locked_only', + help='Only notify if the screen is locked') shell_integration_parser.add_argument( '-f', '--foreground-too', diff --git a/ntfy/screensaver.py b/ntfy/screensaver.py new file mode 100644 index 0000000..3a321b2 --- /dev/null +++ b/ntfy/screensaver.py @@ -0,0 +1,106 @@ +from shlex import split +from subprocess import check_output, check_call, CalledProcessError, PIPE + +# some adapted from +# https://github.com/mtorromeo/xdg-utils/blob/master/scripts/xdg-screensaver.in#L540 + + +def xscreensaver_detect(): + try: + check_call(split('pgrep xscreensaver'), stdout=PIPE) + except (CalledProcessError, OSError): + return False + else: + return True + + +def xscreensaver_is_locked(): + return 'screen locked' in check_output(split('xscreensaver-command -time')) + + +def lightlocker_detect(): + try: + check_call(split('pgrep light-locker'), stdout=PIPE) + except (CalledProcessError, OSError): + return False + else: + return True + + +def lightlocker_is_active(): + return 'The screensaver is active' in check_output(split( + 'light-locker-command -q')) + + +def gnomescreensaver_detect(): + try: + import dbus + except ImportError: + return False + bus = dbus.SessionBus() + dbus_obj = bus.get_object('org.freedesktop.DBus', + '/org/freedesktop/DBus') + dbus_iface = dbus.Interface(dbus_obj, + dbus_interface='org.freedesktop.DBus') + try: + dbus_iface.GetNameOwner('org.gnome.ScreenSaver') + except dbus.DBusException as e: + if e.get_dbus_name() == 'org.freedesktop.DBus.Error.NameHasNoOwner': + return False + else: + raise e + else: + return True + + +def gnomescreensaver_is_locked(): + import dbus + bus = dbus.SessionBus() + dbus_obj = bus.get_object('org.gnome.ScreenSaver', + '/org/gnome/ScreenSaver') + dbus_iface = dbus.Interface(dbus_obj, + dbus_interface='org.gnome.ScreenSaver') + return bool(dbus_iface.GetActive()) + + +def matescreensaver_detect(): + try: + import dbus + except ImportError: + return False + bus = dbus.SessionBus() + dbus_obj = bus.get_object('org.freedesktop.DBus', + '/org/freedesktop/DBus') + dbus_iface = dbus.Interface(dbus_obj, + dbus_interface='org.freedesktop.DBus') + try: + dbus_iface.GetNameOwner('org.mate.ScreenSaver') + except dbus.DBusException as e: + if e.get_dbus_name() == 'org.freedesktop.DBus.Error.NameHasNoOwner': + return False + else: + raise e + else: + return True + + +def matescreensaver_is_locked(): + import dbus + bus = dbus.SessionBus() + dbus_obj = bus.get_object('org.mate.ScreenSaver', + '/org/mate/ScreenSaver') + dbus_iface = dbus.Interface(dbus_obj, + dbus_interface='org.mate.ScreenSaver') + return bool(dbus_iface.GetActive()) + + +def is_locked(): + if xscreensaver_detect(): + return xscreensaver_is_locked() + if lightlocker_detect(): + return lightlocker_is_active() + if gnomescreensaver_detect(): + return gnomescreensaver_is_locked() + if matescreensaver_detect(): + return matescreensaver_is_locked() + return True diff --git a/tests/test_cli.py b/tests/test_cli.py index ec0758f..7b08b3d 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -24,6 +24,7 @@ def test_default(self, mock_Popen): args.pid = None args.unfocused_only = False args.hide_command = False + args.locked_only = False self.assertEqual(('"true" succeeded in 0:00 minutes', 0), run_cmd(args)) @patch('ntfy.cli.Popen') @@ -36,6 +37,7 @@ def test_emoji(self, mock_Popen): args.no_emoji = False args.unfocused_only = False args.hide_command = False + args.locked_only = False self.assertEqual((':white_check_mark: "true" succeeded in 0:00 minutes', 0), run_cmd(args)) @@ -55,6 +57,7 @@ def test_longerthan(self, mock_Popen): args.pid = None args.unfocused_only = False args.hide_command = False + args.locked_only = False self.assertEqual((None, None), run_cmd(args)) @patch('ntfy.cli.Popen') @@ -66,6 +69,7 @@ def test_failure(self, mock_Popen): args.pid = None args.unfocused_only = False args.hide_command = False + args.locked_only = False self.assertEqual(('"false" failed (code 42) in 0:00 minutes', 42), run_cmd(args)) @patch('ntfy.cli.Popen') @@ -77,6 +81,7 @@ def test_stdout(self, mock_Popen): args.pid = None args.unfocused_only = False args.hide_command = False + args.locked_only = False # not actually used args.stdout = True args.stderr = False @@ -91,6 +96,7 @@ def test_stderr(self, mock_Popen): args.pid = None args.unfocused_only = False args.hide_command = False + args.locked_only = False # not actually used args.stdout = False args.stderr = True @@ -105,6 +111,7 @@ def test_stdout_and_stderr(self, mock_Popen): args.pid = None args.unfocused_only = False args.hide_command = False + args.locked_only = False # not actually used args.stdout = True args.stderr = True @@ -119,6 +126,7 @@ def test_failure_stdout_and_stderr(self, mock_Popen): args.pid = None args.unfocused_only = False args.hide_command = False + args.locked_only = False # not actually used args.stdout = True args.stderr = True @@ -143,6 +151,7 @@ def test_formatter(self): args.longer_than = -1 args.unfocused_only = False args.hide_command = False + args.locked_only = False self.assertEqual(('"true" succeeded in 1:05 minutes', 0), run_cmd(args)) def test_formatter_failure(self): @@ -153,6 +162,7 @@ def test_formatter_failure(self): args.longer_than = -1 args.unfocused_only = False args.hide_command = False + args.locked_only = False self.assertEqual(('"false" failed (code 1) in 0:10 minutes', 1), run_cmd(args)) @@ -186,6 +196,7 @@ def test_watch_pid(self, mock_process): args = MagicMock() args.pid = 1 args.unfocused_only = False + args.locked_only = False self.assertEqual('PID[1]: "cmd" finished in 0:00 minutes', run_cmd(args)[0])