Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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()
logging.info('Pause Safe Eyes due to system idle')
disable_safe_eyes(None)
elif system_idle_time < idle_time and context['state'] == State.STOPPED:
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 interpret_idle_as_break and idle_seconds >= next_break_duration:
# User is idle for break duration and wants to consider it as a break
enable_safe_eyes()
elif idle_seconds < break_interval:
# Credit back the idle time
next_break = next_break_time + idle_period
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()
logging.info('Pause Safe Eyes due to system idle')
disable_safe_eyes(None)
elif system_idle_time < idle_time and context['state'] == State.STOPPED:
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 interpret_idle_as_break and idle_seconds >= next_break_duration:
# User is idle for break duration and wants to consider it as a break
enable_safe_eyes()
elif idle_seconds < break_interval:
# Credit back the idle time
next_break = next_break_time + idle_period
enable_safe_eyes(next_break.timestamp())
else:
# User is idle for more than the time between two breaks
enable_safe_eyes()
def quit(self):
"""
Listen to the tray menu quit action and stop the core, notification and the app itself.
"""
logging.info("Quit Safe Eyes")
self.context['state'] = State.QUIT
self.plugins_manager.stop()
self.safe_eyes_core.stop()
self.plugins_manager.exit()
self.__stop_rpc_server()
self.persist_session()
Gtk.main_quit()
# Exit all threads
os._exit(0)
def __fire_pre_break(self):
"""
Show the notification and start the break after the notification.
"""
self.context['state'] = State.PRE_BREAK
if not self.on_pre_break.fire(self.break_queue.get_break()):
# Plugins wanted to ignore this break
self.__start_next_break()
return
Utility.start_thread(self.__wait_until_prepare)
def start(self):
"""
Start Safe Eyes
"""
if self.config.get('use_rpc_server', True):
self.__start_rpc_server()
if self.safe_eyes_core.has_breaks():
self.active = True
self.context['state'] = State.START
self.plugins_manager.start() # Call the start method of all plugins
self.safe_eyes_core.start()
self.handle_system_suspend()
# This event is fired just before the start of a break
self.on_start_break = EventHook()
# This event is fired at the start of a break
self.start_break = EventHook()
# This event is fired during every count down
self.on_count_down = EventHook()
# This event is fired at the end of a break
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
self.context['state'] = State.WAITING