SafeEyes/safeeyes/model.py

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