450 lines
14 KiB
Python
450 lines
14 KiB
Python
#!/usr/bin/env python
|
|
# 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 <http://www.gnu.org/licenses/>.
|
|
"""
|
|
This module contains the entity classes used by Safe Eyes and its plugins.
|
|
"""
|
|
|
|
import logging
|
|
import random
|
|
from enum import Enum
|
|
from dataclasses import dataclass
|
|
|
|
from packaging.version import parse
|
|
|
|
from safeeyes import utility
|
|
|
|
|
|
class Break:
|
|
"""
|
|
An entity class which represents a break.
|
|
"""
|
|
|
|
def __init__(self, break_type, name, time, duration, image, plugins):
|
|
self.type = break_type
|
|
self.name = name
|
|
self.duration = duration
|
|
self.image = image
|
|
self.plugins = plugins
|
|
self.time = time
|
|
self.next = None
|
|
|
|
def __str__(self):
|
|
return 'Break: {{name: "{}", type: {}, duration: {}}}\n'.format(self.name, self.type, self.duration)
|
|
|
|
def __repr__(self):
|
|
return str(self)
|
|
|
|
def is_long_break(self):
|
|
"""
|
|
Check whether this break is a long break.
|
|
"""
|
|
return self.type == BreakType.LONG_BREAK
|
|
|
|
def is_short_break(self):
|
|
"""
|
|
Check whether this break is a short break.
|
|
"""
|
|
return self.type == BreakType.SHORT_BREAK
|
|
|
|
def plugin_enabled(self, plugin_id, is_plugin_enabled):
|
|
"""
|
|
Check whether this break supports the given plugin.
|
|
"""
|
|
if self.plugins:
|
|
return plugin_id in self.plugins
|
|
else:
|
|
return is_plugin_enabled
|
|
|
|
|
|
class BreakType(Enum):
|
|
"""
|
|
Type of Safe Eyes breaks.
|
|
"""
|
|
SHORT_BREAK = 1
|
|
LONG_BREAK = 2
|
|
|
|
|
|
class BreakQueue:
|
|
def __init__(self, config, context):
|
|
self.context = context
|
|
self.__current_break = None
|
|
self.__current_long = 0
|
|
self.__current_short = 0
|
|
self.__short_break_time = config.get('short_break_interval')
|
|
self.__long_break_time = config.get('long_break_interval')
|
|
self.__is_random_order = config.get('random_order')
|
|
self.__config = config
|
|
|
|
self.__build_longs()
|
|
self.__build_shorts()
|
|
|
|
|
|
# Interface guarantees that short_interval >= 1
|
|
# And that long_interval is a multiple of short_interval
|
|
short_interval = config.get('short_break_interval')
|
|
long_interval = config.get('long_break_interval')
|
|
self.__cycle_len = int(long_interval / short_interval)
|
|
# To count every long break as a cycle in .next() if there are no short breaks
|
|
if self.__short_queue is None:
|
|
self.__cycle_len = 1
|
|
|
|
# Restore the last break from session
|
|
if not self.is_empty():
|
|
last_break = context['session'].get('break')
|
|
if last_break is not None:
|
|
current_break = self.get_break()
|
|
if last_break != current_break.name:
|
|
brk = self.next()
|
|
while brk != current_break and brk.name != last_break:
|
|
brk = self.next()
|
|
|
|
def get_break(self, break_type = None):
|
|
if self.__current_break is None:
|
|
self.__current_break = self.next()
|
|
|
|
if break_type is None or self.__current_break.type == break_type:
|
|
return self.__current_break
|
|
|
|
if break_type == BreakType.LONG_BREAK:
|
|
if self.__long_queue is None:
|
|
return None;
|
|
return self.__long_queue[self.__current_long]
|
|
|
|
if self.__short_queue is None:
|
|
return None;
|
|
return self.__short_queue[self.__current_short]
|
|
|
|
def is_long_break(self):
|
|
return self.__current_break is not None and self.__current_break.type == BreakType.LONG_BREAK
|
|
|
|
def next(self, break_type = None):
|
|
break_obj = None
|
|
shorts = self.__short_queue
|
|
longs = self.__long_queue
|
|
|
|
# Reset break that has just ended
|
|
if self.is_long_break():
|
|
self.__current_break.time = self.__long_break_time
|
|
if self.__current_long == 0 and self.__is_random_order:
|
|
# Shuffle queue
|
|
self.__build_longs()
|
|
elif self.__current_break:
|
|
# Reduce the break time from the next long break (default)
|
|
if longs:
|
|
longs[self.__current_long].time -= shorts[self.__current_short].time
|
|
if self.__current_short == 0 and self.__is_random_order:
|
|
self.__build_shorts()
|
|
|
|
if self.is_empty():
|
|
return None
|
|
|
|
if shorts is None:
|
|
break_obj = self.__next_long()
|
|
elif longs is None:
|
|
break_obj = self.__next_short()
|
|
elif break_type == BreakType.LONG_BREAK or longs[self.__current_long].time <= shorts[self.__current_short].time:
|
|
break_obj = self.__next_long()
|
|
else:
|
|
break_obj = self.__next_short()
|
|
|
|
self.__current_break = break_obj
|
|
self.context['session']['break'] = self.__current_break.name
|
|
|
|
return break_obj
|
|
|
|
def reset(self):
|
|
for break_object in self.__short_queue:
|
|
break_object.time = self.__short_break_time
|
|
|
|
for break_object in self.__long_queue:
|
|
break_object.time = self.__long_break_time
|
|
|
|
def is_empty(self, break_type = None):
|
|
"""
|
|
Check if the given break type is empty or not. If the break_type is None, check for both short and long breaks.
|
|
"""
|
|
if break_type == BreakType.SHORT_BREAK:
|
|
return self.__short_queue is None
|
|
elif break_type == BreakType.LONG_BREAK:
|
|
return self.__long_queue is None
|
|
else:
|
|
return self.__short_queue is None and self.__long_queue is None
|
|
|
|
def __next_short(self):
|
|
longs = self.__long_queue
|
|
shorts = self.__short_queue
|
|
break_obj = shorts[self.__current_short]
|
|
self.context['break_type'] = 'short'
|
|
|
|
# Update the index to next
|
|
self.__current_short = (self.__current_short + 1) % len(shorts)
|
|
|
|
return break_obj
|
|
|
|
def __next_long(self):
|
|
longs = self.__long_queue
|
|
break_obj = longs[self.__current_long]
|
|
self.context['break_type'] = 'long'
|
|
|
|
# Update the index to next
|
|
self.__current_long = (self.__current_long + 1) % len(longs)
|
|
|
|
return break_obj
|
|
|
|
def __build_queue(self, break_type, break_configs, break_time, break_duration):
|
|
"""
|
|
Build a queue of breaks.
|
|
"""
|
|
size = len(break_configs)
|
|
|
|
if 0 == size:
|
|
# No breaks
|
|
return None
|
|
|
|
if self.__is_random_order:
|
|
breaks_order = random.sample(break_configs, size)
|
|
else:
|
|
breaks_order = break_configs
|
|
|
|
queue = [None] * size
|
|
for i, break_config in enumerate(breaks_order):
|
|
name = _(break_config['name'])
|
|
duration = break_config.get('duration', break_duration)
|
|
image = break_config.get('image')
|
|
plugins = break_config.get('plugins', None)
|
|
interval = break_config.get('interval', break_time)
|
|
|
|
# Validate time value
|
|
if not isinstance(duration, int) or duration <= 0:
|
|
logging.error('Invalid break duration in: ' +
|
|
str(break_config))
|
|
continue
|
|
|
|
break_obj = Break(break_type, name, interval,
|
|
duration, image, plugins)
|
|
queue[i] = break_obj
|
|
|
|
return queue
|
|
|
|
def __build_shorts(self):
|
|
self.__short_queue = self.__build_queue(BreakType.SHORT_BREAK,
|
|
self.__config.get('short_breaks'),
|
|
self.__short_break_time,
|
|
self.__config.get('short_break_duration'))
|
|
|
|
def __build_longs(self):
|
|
self.__long_queue = self.__build_queue(BreakType.LONG_BREAK,
|
|
self.__config.get('long_breaks'),
|
|
self.__long_break_time,
|
|
self.__config.get('long_break_duration'))
|
|
|
|
|
|
|
|
class State(Enum):
|
|
"""
|
|
Possible states of Safe Eyes.
|
|
"""
|
|
START = 0, # Starting scheduler
|
|
WAITING = 1, # User is working (waiting for next break)
|
|
PRE_BREAK = 2, # Preparing for break
|
|
BREAK = 3, # Break
|
|
STOPPED = 4, # Disabled
|
|
QUIT = 5, # Quitting
|
|
RESTING = 6 # Resting (natural break)
|
|
|
|
|
|
class EventHook:
|
|
"""
|
|
Hook to attach and detach listeners to system events.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.__handlers = []
|
|
|
|
def __iadd__(self, handler):
|
|
self.__handlers.append(handler)
|
|
return self
|
|
|
|
def __isub__(self, handler):
|
|
self.__handlers.remove(handler)
|
|
return self
|
|
|
|
def fire(self, *args, **keywargs):
|
|
"""
|
|
Fire all listeners attached with.
|
|
"""
|
|
for handler in self.__handlers:
|
|
if not handler(*args, **keywargs):
|
|
return False
|
|
return True
|
|
|
|
|
|
class Config:
|
|
"""
|
|
The configuration of Safe Eyes.
|
|
"""
|
|
|
|
def __init__(self, init=True):
|
|
# Read the config files
|
|
self.__user_config = utility.load_json(utility.CONFIG_FILE_PATH)
|
|
self.__system_config = utility.load_json(
|
|
utility.SYSTEM_CONFIG_FILE_PATH)
|
|
# If there any breaking changes in long_breaks, short_breaks or any other keys, use the __force_upgrade list
|
|
self.__force_upgrade = []
|
|
# self.__force_upgrade = ['long_breaks', 'short_breaks']
|
|
|
|
if init:
|
|
# if create_startup_entry finds a broken autostart symlink, it will repair it
|
|
utility.create_startup_entry(force=False)
|
|
if self.__user_config is None:
|
|
utility.initialize_safeeyes()
|
|
self.__user_config = self.__system_config
|
|
self.save()
|
|
else:
|
|
system_config_version = self.__system_config['meta']['config_version']
|
|
meta_obj = self.__user_config.get('meta', None)
|
|
if meta_obj is None:
|
|
# Corrupted user config
|
|
self.__user_config = self.__system_config
|
|
else:
|
|
user_config_version = str(
|
|
meta_obj.get('config_version', '0.0.0'))
|
|
if parse(user_config_version) != parse(system_config_version):
|
|
# Update the user config
|
|
self.__merge_dictionary(
|
|
self.__user_config, self.__system_config)
|
|
self.__user_config = self.__system_config
|
|
# Update the style sheet
|
|
utility.replace_style_sheet()
|
|
|
|
utility.merge_plugins(self.__user_config)
|
|
self.save()
|
|
|
|
def __merge_dictionary(self, old_dict, new_dict):
|
|
"""
|
|
Merge the dictionaries.
|
|
"""
|
|
for key in new_dict:
|
|
if key == "meta" or key in self.__force_upgrade:
|
|
continue
|
|
if key in old_dict:
|
|
new_value = new_dict[key]
|
|
old_value = old_dict[key]
|
|
if type(new_value) is type(old_value):
|
|
# Both properties have same type
|
|
if isinstance(new_value, dict):
|
|
self.__merge_dictionary(old_value, new_value)
|
|
else:
|
|
new_dict[key] = old_value
|
|
|
|
def clone(self):
|
|
config = Config(init=False)
|
|
return config
|
|
|
|
def save(self):
|
|
"""
|
|
Save the configuration to file.
|
|
"""
|
|
utility.write_json(utility.CONFIG_FILE_PATH, self.__user_config)
|
|
|
|
def get(self, key, default_value=None):
|
|
"""
|
|
Get the value.
|
|
"""
|
|
value = self.__user_config.get(key, default_value)
|
|
if value is None:
|
|
value = self.__system_config.get(key, None)
|
|
return value
|
|
|
|
def set(self, key, value):
|
|
"""
|
|
Set the value.
|
|
"""
|
|
self.__user_config[key] = value
|
|
|
|
def __eq__(self, config):
|
|
return self.__user_config == config.__user_config
|
|
|
|
def __ne__(self, config):
|
|
return self.__user_config != config.__user_config
|
|
|
|
|
|
class TrayAction:
|
|
"""
|
|
Data object wrapping name, icon and action.
|
|
"""
|
|
|
|
def __init__(self, name, icon, action, system_icon):
|
|
self.name = name
|
|
self.__icon = icon
|
|
self.action = action
|
|
self.system_icon = system_icon
|
|
self.__toolbar_buttons = []
|
|
|
|
def get_icon(self):
|
|
if self.system_icon:
|
|
return self.__icon
|
|
else:
|
|
image = utility.load_and_scale_image(self.__icon, 16, 16)
|
|
image.show()
|
|
return image
|
|
|
|
def add_toolbar_button(self, button):
|
|
self.__toolbar_buttons.append(button)
|
|
|
|
def reset(self):
|
|
for button in self.__toolbar_buttons:
|
|
button.hide()
|
|
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)
|
|
|
|
@dataclass
|
|
class PluginDependency:
|
|
message: str
|
|
link: str|None = None
|
|
|
|
class RequiredPluginException(Exception):
|
|
def __init__(self, plugin_id, plugin_name: str, message: str|PluginDependency):
|
|
if isinstance(message, PluginDependency):
|
|
msg = message.message
|
|
else:
|
|
msg = message
|
|
|
|
super().__init__(msg)
|
|
|
|
self.plugin_id = plugin_id
|
|
self.plugin_name = plugin_name
|
|
self.message = message
|
|
|
|
def get_plugin_id(self):
|
|
return self.plugin_id
|
|
|
|
def get_plugin_name(self):
|
|
return self.plugin_name
|
|
|
|
def get_message(self):
|
|
return self.message
|