diff --git a/pyproject.toml b/pyproject.toml index 1b2c0505..822c0ca2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -69,6 +69,7 @@ types = [ ] tests = [ "pytest==8.3.5", + "time-machine==2.16.0", ] [tool.mypy] diff --git a/safeeyes/config/locale/safeeyes.pot b/safeeyes/config/locale/safeeyes.pot index 13371059..2b991175 100644 --- a/safeeyes/config/locale/safeeyes.pot +++ b/safeeyes/config/locale/safeeyes.pot @@ -567,3 +567,7 @@ msgstr "" msgid "seconds" msgstr "" + +#, python-format +msgid "Please install one of the command-line tools: %s" +msgstr "" diff --git a/safeeyes/core.py b/safeeyes/core.py index deac1ee9..4b27c3cf 100644 --- a/safeeyes/core.py +++ b/safeeyes/core.py @@ -20,17 +20,20 @@ import datetime import logging -import threading -import time import typing -from safeeyes import utility +from safeeyes.model import Break from safeeyes.model import BreakType from safeeyes.model import BreakQueue from safeeyes.model import EventHook from safeeyes.model import State from safeeyes.model import Config +import gi + +gi.require_version("GLib", "2.0") +from gi.repository import GLib + class SafeEyesCore: """Core of Safe Eyes runs the scheduler and notifies the breaks.""" @@ -45,6 +48,20 @@ class SafeEyesCore: _break_queue: typing.Optional[BreakQueue] = None + # set while __wait_for is running + _timeout_id: typing.Optional[int] = None + _callback: typing.Optional[typing.Callable[[], None]] = None + + # set while __fire_hook is running + _firing_hook: bool = False + + # set while taking a break + _countdown: typing.Optional[int] = 0 + _taking_break: typing.Optional[Break] = None + + # set to true when a break was requested + _take_break_now: bool = False + def __init__(self, context) -> None: """Create an instance of SafeEyesCore and initialize the variables.""" # This event is fired before for a break @@ -59,8 +76,6 @@ def __init__(self, context) -> None: self.on_stop_break = EventHook() # This event is fired when deciding the next break time self.on_update_next_break = EventHook() - self.waiting_condition = threading.Condition() - self.lock = threading.Lock() self.context = context self.context["skipped"] = False self.context["postponed"] = False @@ -85,32 +100,29 @@ def start(self, next_break_time=-1, reset_breaks=False) -> None: if self._break_queue is None: logging.info("No breaks defined, not starting the core") return - with self.lock: - if not self.running: - logging.info("Start Safe Eyes core") - if reset_breaks: - logging.info("Reset breaks to start from the beginning") - self._break_queue.reset() + if not self.running: + logging.info("Start Safe Eyes core") + if reset_breaks: + logging.info("Reset breaks to start from the beginning") + self._break_queue.reset() - self.running = True - self.scheduled_next_break_timestamp = int(next_break_time) - utility.start_thread(self.__scheduler_job) + self.running = True + self.scheduled_next_break_timestamp = int(next_break_time) + self.__scheduler_job() def stop(self, is_resting=False) -> None: """Stop Safe Eyes if it is running.""" - with self.lock: - if not self.running: - return - - logging.info("Stop Safe Eyes core") - self.paused_time = datetime.datetime.now().timestamp() - # Stop the break thread - self.waiting_condition.acquire() - self.running = False - if self.context["state"] != State.QUIT: - self.context["state"] = State.RESTING if (is_resting) else State.STOPPED - self.waiting_condition.notify_all() - self.waiting_condition.release() + if not self.running: + return + + logging.info("Stop Safe Eyes core") + self.paused_time = datetime.datetime.now().timestamp() + # Stop the break thread + self.running = False + if self.context["state"] != State.QUIT: + self.context["state"] = State.RESTING if (is_resting) else State.STOPPED + + self.__wakeup_scheduler() def skip(self) -> None: """User skipped the break using Skip button.""" @@ -147,7 +159,13 @@ def take_break(self, break_type: typing.Optional[BreakType] = None) -> None: return if not self.context["state"] == State.WAITING: return - utility.start_thread(self.__take_break, break_type=break_type) + + if break_type is not None and self._break_queue.get_break().type != break_type: + self._break_queue.next(break_type) + + self._take_break_now = True + + self.__wakeup_scheduler() def has_breaks(self, break_type: typing.Optional[BreakType] = None) -> bool: """Check whether Safe Eyes has breaks or not. @@ -162,32 +180,6 @@ def has_breaks(self, break_type: typing.Optional[BreakType] = None) -> bool: return not self._break_queue.is_empty(break_type) - def __take_break(self, break_type: typing.Optional[BreakType] = None) -> None: - """Show the next break screen.""" - logging.info("Take a break due to external request") - - if self._break_queue is None: - # This will only be called by self.take_break, which checks this - return - - with self.lock: - if not self.running: - return - - logging.info("Stop the scheduler") - - # Stop the break thread - self.waiting_condition.acquire() - self.running = False - self.waiting_condition.notify_all() - self.waiting_condition.release() - time.sleep(1) # Wait for 1 sec to ensure the scheduler is dead - self.running = True - - if break_type is not None and self._break_queue.get_break().type != break_type: - self._break_queue.next(break_type) - utility.execute_main_thread(self.__fire_start_break) - def __scheduler_job(self) -> None: """Scheduler task to execute during every interval.""" if not self.running: @@ -233,9 +225,7 @@ def __scheduler_job(self) -> None: seconds=time_to_wait ) self.context["state"] = State.WAITING - utility.execute_main_thread( - self.__fire_on_update_next_break, self.scheduled_next_break_time - ) + self.__fire_on_update_next_break(self.scheduled_next_break_time) # Wait for the pre break warning period if self.postpone_unit == "seconds": @@ -243,22 +233,32 @@ def __scheduler_job(self) -> None: else: logging.info("Waiting for %d minutes until next break", (time_to_wait / 60)) - self.__wait_for(time_to_wait) + self.__wait_for(time_to_wait, self.__do_pre_break) + + def __fire_on_update_next_break(self, next_break_time: datetime.datetime) -> None: + """Pass the next break information to the registered listeners.""" + if self._break_queue is None: + # This will only be called by methods which check this + return + self.__fire_hook( + self.on_update_next_break, self._break_queue.get_break(), next_break_time + ) + + def __do_pre_break(self) -> None: + if self._take_break_now: + self._take_break_now = False + logging.info("Take a break due to external request") + self.__do_start_break() + return logging.info("Pre-break waiting is over") if not self.running: # This can be reached if another thread changed running while __wait_for was # blocking - return # type: ignore[unreachable] - utility.execute_main_thread(self.__fire_pre_break) - - def __fire_on_update_next_break(self, next_break_time: datetime.datetime) -> None: - """Pass the next break information to the registered listeners.""" - if self._break_queue is None: - # This will only be called by methods which check this return - self.on_update_next_break.fire(self._break_queue.get_break(), next_break_time) + + self.__fire_pre_break() def __fire_pre_break(self) -> None: """Show the notification and start the break after the notification.""" @@ -266,33 +266,38 @@ def __fire_pre_break(self) -> None: # This will only be called by methods which check this return self.context["state"] = State.PRE_BREAK - if not self.on_pre_break.fire(self._break_queue.get_break()): + proceed = self.__fire_hook(self.on_pre_break, self._break_queue.get_break()) + if not proceed: # Plugins wanted to ignore this break self.__start_next_break() return - utility.start_thread(self.__wait_until_prepare) + + self.__wait_until_prepare() def __wait_until_prepare(self) -> None: logging.info( "Wait for %d seconds before the break", self.pre_break_warning_time ) # Wait for the pre break warning period - self.__wait_for(self.pre_break_warning_time) - if not self.running: - return - utility.execute_main_thread(self.__fire_start_break) + self.__wait_for(self.pre_break_warning_time, self.__do_start_break) def __postpone_break(self) -> None: - self.__wait_for(self.postpone_duration) - utility.execute_main_thread(self.__fire_start_break) + self.__wait_for(self.postpone_duration, self.__do_start_break) + + def __do_start_break(self) -> None: + if self._take_break_now: + # already taking a break now, ignore + self._take_break_now = False - def __fire_start_break(self) -> None: + if not self.running: + return if self._break_queue is None: # This will only be called by methods which check this return break_obj = self._break_queue.get_break() # Show the break screen - if not self.on_start_break.fire(break_obj): + proceed = self.__fire_hook(self.on_start_break, break_obj) + if not proceed: # Plugins want to ignore this break self.__start_next_break() return @@ -310,10 +315,10 @@ def __fire_start_break(self) -> None: ) self.__fire_on_update_next_break(self.scheduled_next_break_time) # Wait in user thread - utility.start_thread(self.__postpone_break) + self.__postpone_break() else: - self.start_break.fire(break_obj) - utility.start_thread(self.__start_break) + self.__fire_hook(self.start_break, break_obj) + self.__start_break() def __start_break(self) -> None: """Start the break screen.""" @@ -322,26 +327,44 @@ def __start_break(self) -> None: return self.context["state"] = State.BREAK break_obj = self._break_queue.get_break() - countdown = break_obj.duration - total_break_time = countdown + self._taking_break = break_obj + self._countdown = break_obj.duration - while ( - countdown + self.__cycle_break_countdown() + + def __cycle_break_countdown(self) -> None: + if self._taking_break is None or self._countdown is None: + raise Exception("countdown running without countdown or break") + + if self._take_break_now: + logging.warning("Break requested while already taking a break") + self._take_break_now = False + + if ( + self._countdown > 0 and self.running and not self.context["skipped"] and not self.context["postponed"] ): + countdown = self._countdown + self._countdown -= 1 + + total_break_time = self._taking_break.duration seconds = total_break_time - countdown - self.on_count_down.fire(countdown, seconds) - time.sleep(1) # Sleep for 1 second - countdown -= 1 - utility.execute_main_thread(self.__fire_stop_break) + self.__fire_hook(self.on_count_down, countdown, seconds) + # Sleep for 1 second + self.__wait_for(1, self.__cycle_break_countdown) + else: + self._countdown = None + self._taking_break = None + + self.__fire_stop_break() def __fire_stop_break(self) -> None: # Loop terminated because of timeout (not skipped) -> Close the break alert if not self.context["skipped"] and not self.context["postponed"]: logging.info("Break is terminated automatically") - self.on_stop_break.fire() + self.__fire_hook(self.on_stop_break) # Reset the skipped flag self.context["skipped"] = False @@ -349,11 +372,72 @@ def __fire_stop_break(self) -> None: self.context["postpone_button_disabled"] = False self.__start_next_break() - def __wait_for(self, duration: int) -> None: + def __wait_for( + self, + duration: int, + callback: typing.Callable[[], None], + ) -> None: """Wait until someone wake up or the timeout happens.""" - self.waiting_condition.acquire() - self.waiting_condition.wait(duration) - self.waiting_condition.release() + if self._callback is not None or self._timeout_id is not None: + raise Exception("this should not be called reentrantly") + + self._callback = callback + self._timeout_id = GLib.timeout_add_seconds(duration, self.__on_wakeup) + + def __on_wakeup(self) -> bool: + if self._callback is None or self._timeout_id is None: + raise Exception("Woken up but no callback") + + callback = self._callback + + self._timeout_id = None + self._callback = None + + callback() + + # This signals that the callback should only be called once + return GLib.SOURCE_REMOVE + + def __fire_hook( + self, + hook: EventHook, + *args, + **kwargs, + ) -> bool: + if self._firing_hook: + raise Exception("this should not be called reentrantly") + + self._firing_hook = True + + proceed = hook.fire(*args, **kwargs) + + self._firing_hook = False + + return proceed + + def __wakeup_scheduler(self) -> None: + if (self._callback is None) != (self._timeout_id is None): + # either both are set or none are set + raise Exception("This should never happen") + + if (self._callback is None) and not self._firing_hook: + # neither is set - not running + raise Exception("trying to queue action while core is not running") + elif (self._callback is not None) and self._firing_hook: + # both are set + raise Exception("This should never happen") + + if self._callback is not None and self._timeout_id is not None: + callback = self._callback + + GLib.source_remove(self._timeout_id) + self._timeout_id = None + self._callback = None + + callback() + elif self._firing_hook: + # plugin is running + pass def __start_next_break(self) -> None: if self._break_queue is None: @@ -364,4 +448,4 @@ def __start_next_break(self) -> None: if self.running: # Schedule the break again - utility.start_thread(self.__scheduler_job) + self.__wait_for(0, self.__scheduler_job) diff --git a/safeeyes/glade/break_screen.glade b/safeeyes/glade/break_screen.glade index 94d63915..1ccd3106 100644 --- a/safeeyes/glade/break_screen.glade +++ b/safeeyes/glade/break_screen.glade @@ -25,9 +25,6 @@ 0 0 - - - 1 1 @@ -151,7 +148,6 @@ toolbar - 0 end start diff --git a/safeeyes/model.py b/safeeyes/model.py index 9d02e70a..fd322036 100644 --- a/safeeyes/model.py +++ b/safeeyes/model.py @@ -474,21 +474,32 @@ def __ne__(self, config): class TrayAction: """Data object wrapping name, icon and action.""" - def __init__(self, name, icon, action, system_icon): + __toolbar_buttons: list[Gtk.Button] + + def __init__( + self, + name: str, + icon: str, + action: typing.Callable, + system_icon: bool, + single_use: bool, + ) -> None: self.name = name self.__icon = icon self.action = action self.system_icon = system_icon self.__toolbar_buttons = [] + self.single_use = single_use - def get_icon(self): - if self.system_icon: - image = Gtk.Image.new_from_icon_name(self.__icon) - return image - else: + def get_icon(self) -> Gtk.Image: + if not self.system_icon: image = utility.load_and_scale_image(self.__icon, 16, 16) - image.show() - return image + if image is not None: + image.show() + return image + + image = Gtk.Image.new_from_icon_name(self.__icon) + return image def add_toolbar_button(self, button): self.__toolbar_buttons.append(button) @@ -499,12 +510,20 @@ def reset(self): self.__toolbar_buttons.clear() @classmethod - def build(cls, name, icon_path, icon_id, action): - image = utility.load_and_scale_image(icon_path, 12, 12) - if image is None: - return TrayAction(name, icon_id, action, True) - else: - return TrayAction(name, icon_path, action, False) + def build( + cls, + name: str, + icon_path: typing.Optional[str], + icon_id: str, + action: typing.Callable, + single_use: bool = True, + ) -> "TrayAction": + if icon_path is not None: + image = utility.load_and_scale_image(icon_path, 12, 12) + if image is not None: + return TrayAction(name, icon_path, action, False, single_use) + + return TrayAction(name, icon_id, action, True, single_use) @dataclass diff --git a/safeeyes/plugin_manager.py b/safeeyes/plugin_manager.py index 35cc4107..d9e27a23 100644 --- a/safeeyes/plugin_manager.py +++ b/safeeyes/plugin_manager.py @@ -24,9 +24,7 @@ |- plugin.py |- icon.png (Optional) -The plugin.py can have following methods but all are optional: - - description() - If a custom description has to be displayed, use this function +The plugin.py can have following lifecycle methods but all are optional: - init(context, safeeyes_config, plugin_config) Initialize the plugin. Will be called after loading and after every changes in configuration @@ -50,6 +48,20 @@ Executes once the plugin.py is loaded as a module - disable() Executes if the plugin is disabled at the runtime by the user + +The plugin.py can additionally have the following methods: + - get_widget_title(break_obj) + Returns title of this plugin's widget on the break screen + If this is used, it must also use get_widget_content to work correctly + - get_widget_content(break_obj) + Returns content of this plugin's widget on the break screen + If this is used, it must also use get_widget_title to work correctly + - get_tray_action(break_obj) -> TrayAction | list[TrayAction] + Display button(s) on the break screen's tray that triggers an action + +This method is unused: + - description() + If a custom description has to be displayed, use this function """ import importlib @@ -58,7 +70,7 @@ import sys from safeeyes import utility -from safeeyes.model import PluginDependency, RequiredPluginException +from safeeyes.model import Break, PluginDependency, RequiredPluginException, TrayAction sys.path.append(os.path.abspath(utility.SYSTEM_PLUGINS_DIR)) sys.path.append(os.path.abspath(utility.USER_PLUGINS_DIR)) @@ -207,15 +219,19 @@ def get_break_screen_widgets(self, break_obj): continue return widget.strip() - def get_break_screen_tray_actions(self, break_obj): + def get_break_screen_tray_actions(self, break_obj: Break) -> list[TrayAction]: """Return Tray Actions.""" actions = [] for plugin in self.__plugins.values(): action = plugin.call_plugin_method_break_obj( "get_tray_action", 1, break_obj ) - if action: + if isinstance(action, TrayAction): actions.append(action) + elif isinstance(action, list): + for a in action: + if isinstance(a, TrayAction): + actions.append(a) return actions diff --git a/safeeyes/plugins/audiblealert/config.json b/safeeyes/plugins/audiblealert/config.json index ea464895..1493fa08 100644 --- a/safeeyes/plugins/audiblealert/config.json +++ b/safeeyes/plugins/audiblealert/config.json @@ -2,11 +2,11 @@ "meta": { "name": "Audible Alert", "description": "Play audible alert before and after breaks", - "version": "0.0.3" + "version": "0.0.4" }, "dependencies": { "python_modules": [], - "shell_commands": ["aplay"], + "shell_commands": [], "operating_systems": [], "desktop_environments": [], "resources": ["on_pre_break.wav", "on_stop_break.wav"] @@ -22,6 +22,14 @@ "label": "Play audible alert after breaks", "type": "BOOL", "default": true + }, + { + "id": "volume", + "label": "Alert volume", + "type": "INT", + "default": 100, + "max": 100, + "min": 0 }], "break_override_allowed": true -} \ No newline at end of file +} diff --git a/safeeyes/plugins/audiblealert/dependency_checker.py b/safeeyes/plugins/audiblealert/dependency_checker.py new file mode 100644 index 00000000..ea357871 --- /dev/null +++ b/safeeyes/plugins/audiblealert/dependency_checker.py @@ -0,0 +1,36 @@ +# Safe Eyes is a utility to remind you to take break frequently +# to protect your eyes from eye strain. + +# Copyright (C) 2025 Mel Dafert + +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. + +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. + +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +from safeeyes import utility +from safeeyes.translations import translate as _ + + +def validate(plugin_config, plugin_settings): + commands = ["ffplay", "pw-play"] + exists = False + for command in commands: + if utility.command_exist(command): + exists = True + break + + if not exists: + return _("Please install one of the command-line tools: %s") % ", ".join( + [f"'{command}'" for command in commands] + ) + else: + return None diff --git a/safeeyes/plugins/audiblealert/plugin.py b/safeeyes/plugins/audiblealert/plugin.py index 23002e10..b6e06927 100644 --- a/safeeyes/plugins/audiblealert/plugin.py +++ b/safeeyes/plugins/audiblealert/plugin.py @@ -26,6 +26,7 @@ context = None pre_break_alert = False post_break_alert = False +volume: int = 100 def play_sound(resource_name): @@ -36,16 +37,34 @@ def play_sound(resource_name): resource_name {string} -- name of the wav file resource """ - logging.info("Playing audible alert %s", resource_name) + global volume + + logging.info("Playing audible alert %s at volume %s%%", resource_name, volume) try: # Open the sound file path = utility.get_resource_path(resource_name) if path is None: return - utility.execute_command("aplay", ["-q", path]) - except BaseException: - logging.error("Failed to play audible alert %s", resource_name) + logging.error("Failed to load resource %s", resource_name) + return + + if utility.command_exist("ffplay"): # ffmpeg + utility.execute_command( + "ffplay", + [ + path, + "-nodisp", + "-nostats", + "-hide_banner", + "-autoexit", + "-volume", + str(volume), + ], + ) + elif utility.command_exist("pw-play"): # pipewire + pwvol = volume / 100 # 0 = silent, 1.0 = 100% volume + utility.execute_command("pw-play", ["--volume", str(pwvol), path]) def init(ctx, safeeyes_config, plugin_config): @@ -53,10 +72,16 @@ def init(ctx, safeeyes_config, plugin_config): global context global pre_break_alert global post_break_alert + global volume logging.debug("Initialize Audible Alert plugin") context = ctx pre_break_alert = plugin_config["pre_break_alert"] post_break_alert = plugin_config["post_break_alert"] + volume = int(plugin_config.get("volume", 100)) + if volume > 100: + volume = 100 + if volume < 0: + volume = 0 def on_pre_break(break_obj): diff --git a/safeeyes/plugins/donotdisturb/dependency_checker.py b/safeeyes/plugins/donotdisturb/dependency_checker.py index 96a00de4..91907f98 100644 --- a/safeeyes/plugins/donotdisturb/dependency_checker.py +++ b/safeeyes/plugins/donotdisturb/dependency_checker.py @@ -23,7 +23,10 @@ def validate(plugin_config, plugin_settings): command = None if utility.IS_WAYLAND: - if utility.DESKTOP_ENVIRONMENT == "gnome": + if ( + utility.DESKTOP_ENVIRONMENT == "gnome" + or utility.DESKTOP_ENVIRONMENT == "kde" + ): return None command = "wlrctl" else: diff --git a/safeeyes/plugins/donotdisturb/plugin.py b/safeeyes/plugins/donotdisturb/plugin.py index 63d234bf..4e1dff91 100644 --- a/safeeyes/plugins/donotdisturb/plugin.py +++ b/safeeyes/plugins/donotdisturb/plugin.py @@ -176,6 +176,32 @@ def is_idle_inhibited_gnome(): return bool(result & 0b1000) +def is_idle_inhibited_kde() -> bool: + """KDE Plasma doesn't work with wlrctl, and there is no way to enumerate + fullscreen windows, but KDE does expose a non-standard Inhibited property on + org.freedesktop.Notifications, which does communicate the Do Not Disturb status + on KDE. + This is also only an approximation, but comes pretty close. + """ + dbus_proxy = Gio.DBusProxy.new_for_bus_sync( + bus_type=Gio.BusType.SESSION, + flags=Gio.DBusProxyFlags.NONE, + info=None, + name="org.freedesktop.Notifications", + object_path="/org/freedesktop/Notifications", + interface_name="org.freedesktop.Notifications", + cancellable=None, + ) + prop = dbus_proxy.get_cached_property("Inhibited") + + if prop is None: + return False + + result = prop.unpack() + + return result + + def _window_class_matches(window_class: str, classes: list) -> bool: return any(map(lambda w: w in classes, window_class.split())) @@ -229,29 +255,30 @@ def _normalize_window_classes(classes_as_str: str): return [w.lower() for w in classes_as_str.split()] -def on_pre_break(break_obj): - """Lifecycle method executes before the pre-break period.""" +def __should_skip_break(pre_break: bool) -> bool: if utility.IS_WAYLAND: if utility.DESKTOP_ENVIRONMENT == "gnome": skip_break = is_idle_inhibited_gnome() + elif utility.DESKTOP_ENVIRONMENT == "kde": + skip_break = is_idle_inhibited_kde() else: - skip_break = is_active_window_skipped_wayland(True) + skip_break = is_active_window_skipped_wayland(pre_break) else: - skip_break = is_active_window_skipped_xorg(True) + skip_break = is_active_window_skipped_xorg(pre_break) if dnd_while_on_battery and not skip_break: skip_break = is_on_battery() + + if skip_break: + logging.info("Skipping break due to donotdisturb") + return skip_break +def on_pre_break(break_obj): + """Lifecycle method executes before the pre-break period.""" + return __should_skip_break(pre_break=True) + + def on_start_break(break_obj): """Lifecycle method executes just before the break.""" - if utility.IS_WAYLAND: - if utility.DESKTOP_ENVIRONMENT == "gnome": - skip_break = is_idle_inhibited_gnome() - else: - skip_break = is_active_window_skipped_wayland(False) - else: - skip_break = is_active_window_skipped_xorg(False) - if dnd_while_on_battery and not skip_break: - skip_break = is_on_battery() - return skip_break + return __should_skip_break(pre_break=False) diff --git a/safeeyes/plugins/screensaver/plugin.py b/safeeyes/plugins/screensaver/plugin.py index 5e75b16c..c8b910b4 100644 --- a/safeeyes/plugins/screensaver/plugin.py +++ b/safeeyes/plugins/screensaver/plugin.py @@ -33,6 +33,7 @@ min_seconds = 0 seconds_passed = 0 tray_icon_path = None +icon_lock_later_path = None def __lock_screen_command(): @@ -102,21 +103,29 @@ def __lock_screen_command(): return None -def __lock_screen(): +def __lock_screen_later(): global user_locked_screen user_locked_screen = True +def __lock_screen_now() -> None: + utility.execute_command(lock_screen_command) + + def init(ctx, safeeyes_config, plugin_config): """Initialize the screensaver plugin.""" global context global lock_screen_command global min_seconds global tray_icon_path + global icon_lock_later_path logging.debug("Initialize Screensaver plugin") context = ctx min_seconds = plugin_config["min_seconds"] tray_icon_path = os.path.join(plugin_config["path"], "resource/lock.png") + icon_lock_later_path = os.path.join( + plugin_config["path"], "resource/rotation-lock-symbolic.svg" + ) if plugin_config["command"]: lock_screen_command = plugin_config["command"].split() else: @@ -147,10 +156,22 @@ def on_stop_break(): min_seconds. """ if user_locked_screen or (lock_screen and seconds_passed >= min_seconds): - utility.execute_command(lock_screen_command) - - -def get_tray_action(break_obj): - return TrayAction.build( - "Lock screen", tray_icon_path, "dialog-password", __lock_screen - ) + __lock_screen_now() + + +def get_tray_action(break_obj) -> list[TrayAction]: + return [ + TrayAction.build( + "Lock screen now", + tray_icon_path, + "system-lock-screen", + __lock_screen_now, + single_use=False, + ), + TrayAction.build( + "Lock screen after break", + icon_lock_later_path, + "dialog-password", + __lock_screen_later, + ), + ] diff --git a/safeeyes/plugins/screensaver/resource/rotation-lock-symbolic.svg b/safeeyes/plugins/screensaver/resource/rotation-lock-symbolic.svg new file mode 100644 index 00000000..d2a86cf1 --- /dev/null +++ b/safeeyes/plugins/screensaver/resource/rotation-lock-symbolic.svg @@ -0,0 +1,51 @@ + + + + + + + + + + diff --git a/safeeyes/safeeyes.py b/safeeyes/safeeyes.py index 56c1e48a..bca9fa16 100644 --- a/safeeyes/safeeyes.py +++ b/safeeyes/safeeyes.py @@ -23,7 +23,6 @@ import atexit import logging import typing -from threading import Timer from importlib import metadata import gi @@ -492,8 +491,7 @@ def restart(self, config, set_active=False): self.active = True if self.active and self.safe_eyes_core.has_breaks(): - # 1 sec delay is required to give enough time for core to be stopped - Timer(1.0, self.safe_eyes_core.start).start() + self.safe_eyes_core.start() self.plugins_manager.start() def enable_safeeyes(self, scheduled_next_break_time=-1, reset_breaks=False): diff --git a/safeeyes/tests/test_core.py b/safeeyes/tests/test_core.py new file mode 100644 index 00000000..96b1a200 --- /dev/null +++ b/safeeyes/tests/test_core.py @@ -0,0 +1,544 @@ +# Safe Eyes is a utility to remind you to take break frequently +# to protect your eyes from eye strain. + +# Copyright (C) 2025 Mel Dafert + +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. + +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. + +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +import datetime +import pytest +import typing + +from safeeyes import core +from safeeyes import model + +from time_machine import TimeMachineFixture + +from unittest import mock + + +class SafeEyesCoreHandle: + callback: typing.Optional[typing.Tuple[typing.Callable, int]] = None + safe_eyes_core: core.SafeEyesCore + time_machine: TimeMachineFixture + + def __init__( + self, + safe_eyes_core: core.SafeEyesCore, + time_machine: TimeMachineFixture, + ): + self.time_machine = time_machine + self.safe_eyes_core = safe_eyes_core + + def timeout_add_seconds(self, duration: int, callback: typing.Callable) -> int: + if self.callback is not None: + raise Exception("only one callback supported. need to make this smarter") + self.callback = (callback, duration) + print(f"callback registered for {callback} and {duration}") + return 1 + + def next(self) -> None: + assert self.callback + + (callback, duration) = self.callback + self.callback = None + self.time_machine.shift(delta=datetime.timedelta(seconds=duration)) + print(f"shift to {datetime.datetime.now()}") + callback() + + +SequentialThreadingFixture: typing.TypeAlias = typing.Callable[ + [core.SafeEyesCore], SafeEyesCoreHandle +] + + +class TestSafeEyesCore: + @pytest.fixture(autouse=True) + def set_time(self, time_machine): + time_machine.move_to( + datetime.datetime.fromisoformat("2024-08-25T13:00:00+00:00"), tick=False + ) + + @pytest.fixture(autouse=True) + def monkeypatch_translations(self, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr( + core, "_", lambda message: "translated!: " + message, raising=False + ) + monkeypatch.setattr( + model, "_", lambda message: "translated!: " + message, raising=False + ) + + @pytest.fixture + def sequential_threading( + self, + monkeypatch: pytest.MonkeyPatch, + time_machine: TimeMachineFixture, + ) -> typing.Generator[SequentialThreadingFixture]: + """This fixture allows stopping threads at any point. + + It is hard-coded for SafeEyesCore, the handle class returned by the fixture must + be initialized with a SafeEyesCore instance to be patched. + With this, all sleeping/blocking/thread starting calls inside SafeEyesCore are + intercepted, and paused. + Additionally, all threads inside SafeEyesCore run sequentially. + The test code can use the next() method to unpause the thread, + which will run until the next sleeping/blocking/thread starting call, + after which it needs to be woken up using next() again. + The next() method blocks the test code while the thread is running. + """ + # executes instantly, on the same thread + # no need to switch threads, as we don't use any gtk things + handle: typing.Optional["SafeEyesCoreHandle"] = None + + def timeout_add_seconds(duration, callback) -> int: + if not handle: + raise Exception("handle must be initialized before first sleep call") + return handle.timeout_add_seconds(duration, callback) + + def source_remove(source_id: int) -> None: + pass + + monkeypatch.setattr(core.GLib, "timeout_add_seconds", timeout_add_seconds) + monkeypatch.setattr(core.GLib, "source_remove", source_remove) + + def create_handle(safe_eyes_core: core.SafeEyesCore) -> SafeEyesCoreHandle: + nonlocal time_machine + nonlocal handle + if handle: + raise Exception("only one handle is allowed per test call") + + handle = SafeEyesCoreHandle(safe_eyes_core, time_machine) + + return handle + + yield create_handle + + def run_next_break( + self, + sequential_threading_handle: SafeEyesCoreHandle, + time_machine: TimeMachineFixture, + safe_eyes_core: core.SafeEyesCore, + context, + break_duration: int, + break_name_translated: str, + initial: bool = False, + ): + """Run one entire cycle of safe_eyes_core. + + If initial is True, it must not be started yet. + If initial is False, it must be in the state where __scheduler_job is about to + be called again. + This means it is in the BREAK state, but the break has ended and on_stop_break + was already called. + """ + on_update_next_break = mock.Mock() + on_pre_break = mock.Mock(return_value=True) + on_start_break = mock.Mock(return_value=True) + start_break = mock.Mock() + on_count_down = mock.Mock() + on_stop_break = mock.Mock() + + safe_eyes_core.on_update_next_break += on_update_next_break + safe_eyes_core.on_pre_break += on_pre_break + safe_eyes_core.on_start_break += on_start_break + safe_eyes_core.start_break += start_break + safe_eyes_core.on_count_down += on_count_down + safe_eyes_core.on_stop_break += on_stop_break + + if initial: + safe_eyes_core.start() + else: + assert context["state"] == model.State.BREAK + + sequential_threading_handle.next() + + assert context["state"] == model.State.WAITING + + on_update_next_break.assert_called_once() + assert isinstance(on_update_next_break.call_args[0][0], model.Break) + assert on_update_next_break.call_args[0][0].name == break_name_translated + on_update_next_break.reset_mock() + + # continue after condvar + sequential_threading_handle.next() + # end of __scheduler_job + + assert context["state"] == model.State.PRE_BREAK + + on_pre_break.assert_called_once() + assert isinstance(on_pre_break.call_args[0][0], model.Break) + assert on_pre_break.call_args[0][0].name == break_name_translated + on_pre_break.reset_mock() + + # start __wait_until_prepare + # first sleep in __start_break + sequential_threading_handle.next() + + assert context["state"] == model.State.BREAK + + on_start_break.assert_called_once() + assert isinstance(on_start_break.call_args[0][0], model.Break) + assert on_start_break.call_args[0][0].name == break_name_translated + on_start_break.reset_mock() + + start_break.assert_called_once() + assert isinstance(start_break.call_args[0][0], model.Break) + assert start_break.call_args[0][0].name == break_name_translated + start_break.reset_mock() + + assert context["state"] == model.State.BREAK + + # continue sleep in __start_break + for i in range(break_duration - 1): + sequential_threading_handle.next() + + assert context["state"] == model.State.BREAK + + sequential_threading_handle.next() + # end of __start_break + + on_count_down.assert_called() + assert on_count_down.call_count == break_duration + on_count_down.reset_mock() + + on_stop_break.assert_called() + assert on_stop_break.call_count == 1 + on_stop_break.reset_mock() + + assert context["state"] == model.State.BREAK + + def assert_datetime(self, string: str): + if not string.endswith("+00:00"): + string += "+00:00" + assert datetime.datetime.now( + datetime.timezone.utc + ) == datetime.datetime.fromisoformat(string) + + def test_start_empty(self, sequential_threading: SequentialThreadingFixture): + context: dict[str, typing.Any] = {} + config = model.Config( + user_config={ + "short_breaks": [], + "long_breaks": [], + "short_break_interval": 15, + "long_break_interval": 75, + "long_break_duration": 60, + "short_break_duration": 15, + "random_order": False, + "postpone_duration": 5, + }, + system_config={}, + ) + on_update_next_break = mock.Mock() + safe_eyes_core = core.SafeEyesCore(context) + safe_eyes_core.on_update_next_break += mock + + safe_eyes_core.initialize(config) + + safe_eyes_core.start() + safe_eyes_core.stop() + + on_update_next_break.assert_not_called() + + def test_start(self, sequential_threading: SequentialThreadingFixture): + context: dict[str, typing.Any] = { + "session": {}, + } + config = model.Config( + user_config={ + "short_breaks": [ + {"name": "break 1"}, + {"name": "break 2"}, + {"name": "break 3"}, + {"name": "break 4"}, + ], + "long_breaks": [ + {"name": "long break 1"}, + {"name": "long break 2"}, + {"name": "long break 3"}, + ], + "short_break_interval": 15, + "long_break_interval": 75, + "long_break_duration": 60, + "short_break_duration": 15, + "random_order": False, + "postpone_duration": 5, + }, + system_config={}, + ) + on_update_next_break = mock.Mock() + safe_eyes_core = core.SafeEyesCore(context) + safe_eyes_core.on_update_next_break += on_update_next_break + + safe_eyes_core.initialize(config) + + sequential_threading_handle = sequential_threading(safe_eyes_core) + + safe_eyes_core.start() + + assert context["state"] == model.State.WAITING + + on_update_next_break.assert_called_once() + assert isinstance(on_update_next_break.call_args[0][0], model.Break) + assert on_update_next_break.call_args[0][0].name == "translated!: break 1" + on_update_next_break.reset_mock() + + # wait for end of __scheduler_job - we cannot stop while waiting on the condvar + # this just moves us into waiting for __wait_until_prepare to start + sequential_threading_handle.next() + + safe_eyes_core.stop() + assert context["state"] == model.State.STOPPED + + def test_full_run_with_defaults( + self, + sequential_threading: SequentialThreadingFixture, + time_machine: TimeMachineFixture, + ): + context: dict[str, typing.Any] = { + "session": {}, + } + short_break_duration = 15 # seconds + short_break_interval = 15 # minutes + pre_break_warning_time = 10 # seconds + long_break_duration = 60 # seconds + long_break_interval = 75 # minutes + config = model.Config( + user_config={ + "short_breaks": [ + {"name": "break 1"}, + {"name": "break 2"}, + {"name": "break 3"}, + {"name": "break 4"}, + ], + "long_breaks": [ + {"name": "long break 1"}, + {"name": "long break 2"}, + {"name": "long break 3"}, + ], + "short_break_interval": short_break_interval, + "long_break_interval": long_break_interval, + "long_break_duration": long_break_duration, + "short_break_duration": short_break_duration, + "pre_break_warning_time": pre_break_warning_time, + "random_order": False, + "postpone_duration": 5, + }, + system_config={}, + ) + + self.assert_datetime("2024-08-25T13:00:00") + + safe_eyes_core = core.SafeEyesCore(context) + + sequential_threading_handle = sequential_threading(safe_eyes_core) + + safe_eyes_core.initialize(config) + + self.run_next_break( + sequential_threading_handle, + time_machine, + safe_eyes_core, + context, + short_break_duration, + "translated!: break 1", + initial=True, + ) + + # Time passed: 15min 25s + # 15min short_break_interval, 10 seconds pre_break_warning_time, + # 15 seconds short_break_duration + self.assert_datetime("2024-08-25T13:15:25") + + self.run_next_break( + sequential_threading_handle, + time_machine, + safe_eyes_core, + context, + short_break_duration, + "translated!: break 2", + ) + + self.assert_datetime("2024-08-25T13:30:50") + + self.run_next_break( + sequential_threading_handle, + time_machine, + safe_eyes_core, + context, + short_break_duration, + "translated!: break 3", + ) + + self.assert_datetime("2024-08-25T13:46:15") + + self.run_next_break( + sequential_threading_handle, + time_machine, + safe_eyes_core, + context, + short_break_duration, + "translated!: break 4", + ) + + self.assert_datetime("2024-08-25T14:01:40") + + self.run_next_break( + sequential_threading_handle, + time_machine, + safe_eyes_core, + context, + long_break_duration, + "translated!: long break 1", + ) + + # Time passed: 16min 10s + # 15min short_break_interval (from previous, as long_break_interval must be + # multiple) + # 10 seconds pre_break_warning_time, 1 minute long_break_duration + self.assert_datetime("2024-08-25T14:17:50") + + self.run_next_break( + sequential_threading_handle, + time_machine, + safe_eyes_core, + context, + short_break_duration, + "translated!: break 1", + ) + + # Time passed: 15min 25s + # 15min short_break_interval, 10 seconds pre_break_warning_time, + # 15 seconds short_break_duration + self.assert_datetime("2024-08-25T14:33:15") + + safe_eyes_core.stop() + + assert context["state"] == model.State.STOPPED + + def test_long_duration_is_bigger_than_short_interval( + self, + sequential_threading: SequentialThreadingFixture, + time_machine: TimeMachineFixture, + ): + """Example taken from https://github.com/slgobinath/SafeEyes/issues/640.""" + context: dict[str, typing.Any] = { + "session": {}, + } + short_break_duration = 300 # seconds = 5min + short_break_interval = 25 # minutes + pre_break_warning_time = 10 # seconds + long_break_duration = 1800 # seconds = 30min + long_break_interval = 100 # minutes + config = model.Config( + user_config={ + "short_breaks": [ + {"name": "break 1"}, + {"name": "break 2"}, + {"name": "break 3"}, + {"name": "break 4"}, + ], + "long_breaks": [ + {"name": "long break 1"}, + {"name": "long break 2"}, + {"name": "long break 3"}, + ], + "short_break_interval": short_break_interval, + "long_break_interval": long_break_interval, + "long_break_duration": long_break_duration, + "short_break_duration": short_break_duration, + "pre_break_warning_time": pre_break_warning_time, + "random_order": False, + "postpone_duration": 5, + }, + system_config={}, + ) + + self.assert_datetime("2024-08-25T13:00:00") + + safe_eyes_core = core.SafeEyesCore(context) + + sequential_threading_handle = sequential_threading(safe_eyes_core) + + safe_eyes_core.initialize(config) + + self.run_next_break( + sequential_threading_handle, + time_machine, + safe_eyes_core, + context, + short_break_duration, + "translated!: break 1", + initial=True, + ) + + # Time passed: 30m 10s + # 25min short_break_interval, 10 seconds pre_break_warning_time, + # 5 minutes short_break_duration + self.assert_datetime("2024-08-25T13:30:10") + + self.run_next_break( + sequential_threading_handle, + time_machine, + safe_eyes_core, + context, + short_break_duration, + "translated!: break 2", + ) + + self.assert_datetime("2024-08-25T14:00:20") + + self.run_next_break( + sequential_threading_handle, + time_machine, + safe_eyes_core, + context, + short_break_duration, + "translated!: break 3", + ) + + self.assert_datetime("2024-08-25T14:30:30") + + self.run_next_break( + sequential_threading_handle, + time_machine, + safe_eyes_core, + context, + long_break_duration, + "translated!: long break 1", + ) + + # Time passed: 55min 10s + # 25min short_break_interval (from previous, as long_break_interval must be + # multiple) + # 10 seconds pre_break_warning_time, 30 minute long_break_duration + self.assert_datetime("2024-08-25T15:25:40") + + self.run_next_break( + sequential_threading_handle, + time_machine, + safe_eyes_core, + context, + short_break_duration, + "translated!: break 4", + ) + + # Time passed: 30m 10s + # 15min short_break_interval, 10 seconds pre_break_warning_time, + # 15 seconds short_break_duration + self.assert_datetime("2024-08-25T15:55:50") + + safe_eyes_core.stop() + + assert context["state"] == model.State.STOPPED diff --git a/safeeyes/ui/break_screen.py b/safeeyes/ui/break_screen.py index aeacff15..158adc42 100644 --- a/safeeyes/ui/break_screen.py +++ b/safeeyes/ui/break_screen.py @@ -23,6 +23,7 @@ import gi from safeeyes import utility +from safeeyes.model import TrayAction from safeeyes.translations import translate as _ import Xlib from Xlib.display import Display @@ -139,13 +140,14 @@ def close(self): # Destroy other windows if exists GLib.idle_add(lambda: self.__destroy_all_screens()) - def __tray_action(self, button, tray_action): + def __tray_action(self, button, tray_action: TrayAction): """Tray action handler. - Hides all toolbar buttons for this action and call the action - provided by the plugin. + Hides all toolbar buttons for this action, if it is single use, + and call the action provided by the plugin. """ - tray_action.reset() + if tray_action.single_use: + tray_action.reset() tray_action.action() def __show_break_screen(self, message, image_path, widget, tray_actions): diff --git a/safeeyes/utility.py b/safeeyes/utility.py index 28875471..c8908438 100644 --- a/safeeyes/utility.py +++ b/safeeyes/utility.py @@ -31,6 +31,7 @@ import shutil import subprocess import threading +import typing from logging.handlers import RotatingFileHandler from pathlib import Path @@ -732,7 +733,9 @@ def create_gtk_builder(glade_file): return builder -def load_and_scale_image(path, width, height): +def load_and_scale_image( + path: str, width: int, height: int +) -> typing.Optional[Gtk.Image]: if not os.path.isfile(path): return None pixbuf = GdkPixbuf.Pixbuf.new_from_file_at_scale(