# Safe Eyes is a utility to remind you to take break frequently # to protect your eyes from eye strain. # Copyright (C) 2017 Gobinath # 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 logging import subprocess import threading import re import os from safeeyes import utility from safeeyes.model import State """ Safe Eyes smart pause plugin """ context = None idle_condition = threading.Condition() lock = threading.Lock() active = False idle_time = 0 enable_safeeyes = None disable_safeeyes = None smart_pause_activated = False idle_start_time = None next_break_time = None next_break_duration = 0 short_break_interval = 0 waiting_time = 2 is_wayland_and_gnome = False use_swayidle = False swayidle_process = None swayidle_lock = threading.Lock() swayidle_idle = 0 swayidle_active = 0 def __swayidle_running(): return (swayidle_process is not None and swayidle_process.poll() is None) def __start_swayidle_monitor(): global swayidle_process global swayidle_start global swayidle_idle global swayidle_active logging.debug('Starting swayidle subprocess') swayidle_process = subprocess.Popen([ "swayidle", "timeout", "1", "date +S%s", "resume", "date +R%s" ], stdout=subprocess.PIPE, bufsize=1, universal_newlines=True, encoding='utf-8') for line in swayidle_process.stdout: with swayidle_lock: typ = line[0] timestamp = int(line[1:]) if typ == 'S': swayidle_idle = timestamp elif typ == 'R': swayidle_active = timestamp def __stop_swayidle_monitor(): if __swayidle_running(): logging.debug('Stopping swayidle subprocess') swayidle_process.terminate() def __swayidle_idle_time(): with swayidle_lock: if not __swayidle_running(): utility.start_thread(__start_swayidle_monitor) # Idle more recently than active, meaning idle time isn't stale. if swayidle_idle > swayidle_active: idle_time = int(datetime.datetime.now().timestamp()) - swayidle_idle return idle_time return 0 def __gnome_wayland_idle_time(): """ Determine system idle time in seconds, specifically for gnome with wayland. If there's a failure, return 0. https://unix.stackexchange.com/a/492328/222290 """ try: output = subprocess.check_output([ 'dbus-send', '--print-reply', '--dest=org.gnome.Mutter.IdleMonitor', '/org/gnome/Mutter/IdleMonitor/Core', 'org.gnome.Mutter.IdleMonitor.GetIdletime' ]) return int(re.search(rb'\d+$', output).group(0)) / 1000 except BaseException as e: logging.warning("Failed to get system idle time for gnome/wayland.") logging.warning(str(e)) return 0 def __system_idle_time(): """ Get system idle time in minutes. Return the idle time if xprintidle is available, otherwise return 0. """ try: if is_wayland_and_gnome: return __gnome_wayland_idle_time() elif use_swayidle: return __swayidle_idle_time() # Convert to seconds return int(subprocess.check_output(['xprintidle']).decode('utf-8')) / 1000 except BaseException: return 0 def __is_active(): """ Thread safe function to see if this plugin is active or not. """ is_active = False with lock: is_active = active return is_active def __set_active(is_active): """ Thread safe function to change the state of the plugin. """ global active with lock: active = is_active def init(ctx, safeeyes_config, plugin_config): """ Initialize the plugin. """ global context global enable_safeeyes global disable_safeeyes global postpone global idle_time global short_break_interval global long_break_duration global waiting_time global postpone_if_active global is_wayland_and_gnome global use_swayidle logging.debug('Initialize Smart Pause plugin') context = ctx enable_safeeyes = context['api']['enable_safeeyes'] disable_safeeyes = context['api']['disable_safeeyes'] postpone = context['api']['postpone'] idle_time = plugin_config['idle_time'] postpone_if_active = plugin_config['postpone_if_active'] short_break_interval = safeeyes_config.get( 'short_break_interval') * 60 # Convert to seconds long_break_duration = safeeyes_config.get('long_break_duration') waiting_time = min(2, idle_time) # If idle time is 1 sec, wait only 1 sec is_wayland_and_gnome = context['desktop'] == 'gnome' and context['is_wayland'] use_swayidle = context['desktop'] == 'sway' def __start_idle_monitor(): """ Continuously check the system idle time and pause/resume Safe Eyes based on it. """ global smart_pause_activated global idle_start_time while __is_active(): # Wait for waiting_time seconds idle_condition.acquire() idle_condition.wait(waiting_time) idle_condition.release() if __is_active(): # Get the system idle time system_idle_time = __system_idle_time() if system_idle_time >= idle_time and context['state'] == State.WAITING: smart_pause_activated = True idle_start_time = datetime.datetime.now() - datetime.timedelta(seconds=system_idle_time) logging.info('Pause Safe Eyes due to system idle') disable_safeeyes(None, True) elif system_idle_time < idle_time and context['state'] == State.RESTING and idle_start_time is not None: logging.info('Resume Safe Eyes due to user activity') smart_pause_activated = False idle_period = (datetime.datetime.now() - idle_start_time) idle_seconds = idle_period.total_seconds() context['idle_period'] = idle_seconds if idle_seconds < short_break_interval: # Credit back the idle time if next_break_time is not None: # This method runs in a thread since the start. # It may run before next_break is initialized in the update_next_break method next_break = next_break_time + idle_period enable_safeeyes(next_break.timestamp()) else: enable_safeeyes() else: # User is idle for more than the time between two breaks enable_safeeyes() def on_start(): """ Start a thread to continuously call xprintidle. """ global active if not __is_active(): # If SmartPause is already started, do not start it again logging.debug('Start Smart Pause plugin') __set_active(True) utility.start_thread(__start_idle_monitor) def on_stop(): """ Stop the thread from continuously calling xprintidle. """ global active global smart_pause_activated if smart_pause_activated: # Safe Eyes is stopped due to system idle smart_pause_activated = False return logging.debug('Stop Smart Pause plugin') if use_swayidle: __stop_swayidle_monitor() __set_active(False) idle_condition.acquire() idle_condition.notify_all() idle_condition.release() def update_next_break(break_obj, dateTime): """ Update the next break time. """ global next_break_time global next_break_duration next_break_time = dateTime next_break_duration = break_obj.duration def on_start_break(break_obj): """ Lifecycle method executes just before the break. """ if postpone_if_active: # Postpone this break if the user is active system_idle_time = __system_idle_time() if system_idle_time < 2: postpone(2) # Postpone for 2 seconds def disable(): """ SmartPause plugin was active earlier but now user has disabled it. """ # Remove the idle_period context.pop('idle_period', None)