miracle091/SafeEyes
miracle091
/
SafeEyes
mirror of https://github.com/slgobinath/SafeEyes.git
1
0
Fork 0
Code Issues Releases Wiki Activity

Move the rpc server as a plugin

This commit is contained in:
Gobinath 2021-06-25 08:07:18 -04:00
parent 2fb22bbc2d
commit defe27c577
27 changed files with 777 additions and 484 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

100
safeeyes/__main__.py
View File

@ -21,82 +21,25 @@ Safe Eyes is a utility to remind you to take break frequently to protect your ey
"""
import argparse
import gettext
import locale
import logging
import signal
import sys
from threading import Timer
import gi
import psutil
from safeeyes import SAFE_EYES_VERSION
from safeeyes import utility
from safeeyes.config import Config
from safeeyes.rpc import RPCClient
from safeeyes.safeeyes import SafeEyes
from safeeyes.util import locale
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
gettext.install('safeeyes', utility.LOCALE_PATH)
def __running():
"""
Check if SafeEyes is already running.
"""
process_count = 0
for proc in psutil.process_iter():
if not proc.cmdline:
continue
try:
# Check if safeeyes is in process arguments
if callable(proc.cmdline):
# Latest psutil has cmdline function
cmd_line = proc.cmdline()
else:
# In older versions cmdline was a list object
cmd_line = proc.cmdline
if ('python3' in cmd_line[0] or 'python' in cmd_line[0]) and (
'safeeyes' in cmd_line[1] or 'safeeyes' in cmd_line):
process_count += 1
if process_count > 1:
return True
# Ignore if process does not exist or does not have command line args
except (IndexError, psutil.NoSuchProcess):
pass
return False
def __evaluate_arguments(args, safe_eyes):
"""
Evaluate the arguments and execute the operations.
"""
# if args.about:
# utility.execute_main_thread(safe_eyes.show_about)
# elif args.disable:
# utility.execute_main_thread(safe_eyes.disable_safeeyes)
# elif args.enable:
# utility.execute_main_thread(safe_eyes.enable_safeeyes)
# elif args.settings:
# utility.execute_main_thread(safe_eyes.show_settings)
# elif args.take_break:
# utility.execute_main_thread(safe_eyes.take_break)
pass
def main():
"""
Start the Safe Eyes.
"""
system_locale = gettext.translation('safeeyes', localedir=utility.LOCALE_PATH,
languages=[utility.system_locale(), 'en_US'], fallback=True)
system_locale.install()
# locale.bindtextdomain is required for Glade files
# gettext.bindtextdomain(gettext.textdomain(), Utility.LOCALE_PATH)
locale.bindtextdomain('safeeyes', utility.LOCALE_PATH)
system_locale = locale.init_locale()
parser = argparse.ArgumentParser(prog='safeeyes', description=_('description'))
group = parser.add_mutually_exclusive_group()
@ -116,44 +59,9 @@ def main():
# Initialize the logging
utility.intialize_logging(args.debug)
utility.initialize_platform()
config = Config()
if __running():
logging.info("Safe Eyes is already running")
if not config.get("use_rpc_server", True):
# RPC sever is disabled
print(_('Safe Eyes is running without an RPC server. Turn it on to use command-line arguments.'))
sys.exit(0)
return
rpc_client = RPCClient(config.get('rpc_port'))
if args.about:
rpc_client.show_about()
elif args.disable:
rpc_client.disable_safeeyes()
elif args.enable:
rpc_client.enable_safeeyes()
elif args.settings:
rpc_client.show_settings()
elif args.take_break:
rpc_client.take_break()
elif args.status:
print(rpc_client.status())
elif args.quit:
rpc_client.quit()
else:
# Default behavior is opening settings
rpc_client.show_settings()
sys.exit(0)
else:
if args.status:
print(_('Safe Eyes is not running'))
sys.exit(0)
elif not args.quit:
logging.info("Starting Safe Eyes")
safe_eyes = SafeEyes(system_locale, config)
safe_eyes.start()
Timer(1.0, lambda: __evaluate_arguments(args, safe_eyes)).start()
Gtk.main()
safe_eyes = SafeEyes(system_locale)
safe_eyes.run(args)
if __name__ == '__main__':

71
safeeyes/args.py Normal file
View File

@ -0,0 +1,71 @@
# Safe Eyes is a utility to remind you to take break frequently
# to protect your eyes from eye strain.
# Copyright (C) 2021 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/>.
from threading import Timer
from safeeyes.context import Context
from safeeyes.plugin_utils.manager import PluginManager
from safeeyes.util.locale import _
class Arguments:
def __init__(self, context: Context, plugin_manager: PluginManager, args):
self.__context: Context = context
self.__args = args
self.__plugin_manager: PluginManager = plugin_manager
def execute_on_local_instance(self) -> None:
"""
Evaluate the arguments and execute the operations.
"""
Timer(1.0, lambda: self.__execute_on_local_instance()).start()
def __execute_on_local_instance(self) -> None:
if self.__args.about:
self.__context.window_api.show_about()
elif self.__args.disable:
self.__context.core_api.stop()
elif self.__args.enable:
self.__context.core_api.start()
elif self.__args.settings:
self.__context.window_api.show_settings()
elif self.__args.take_break:
self.__context.break_api.take_break()
def execute_on_remote_instance(self) -> None:
rpc_plugin = self.__plugin_manager.get_plugin("rpcserver")
if rpc_plugin:
if self.__args.about:
rpc_plugin.execute_if_exists("show_about")
elif self.__args.disable:
rpc_plugin.execute_if_exists("disable_safe_eyes")
elif self.__args.enable:
rpc_plugin.execute_if_exists("enable_safe_eyes")
elif self.__args.settings:
rpc_plugin.execute_if_exists("show_settings")
elif self.__args.take_break:
rpc_plugin.execute_if_exists("take_break")
elif self.__args.status:
print(rpc_plugin.execute_if_exists("get_status"))
elif self.__args.quit:
rpc_plugin.execute_if_exists("quit_safe_eyes")
else:
# Default behavior is opening settings
rpc_plugin.execute_if_exists("show_settings")
else:
print(_("RPC Server plugin is disabled. Without this plugin, command line arguments cannot be processed"))

16
safeeyes/config/locale/safeeyes.pot
View File

@ -346,6 +346,22 @@ msgstr ""
msgid "Ready for a long break in %s seconds"
msgstr ""
# plugin/rpc
msgid "Port"
msgstr ""
# plugin/rpc
msgid "RPC server to receive command-line arguments"
msgstr ""
# plugin/rpc
msgid "Maximum adjacent ports to try"
msgstr ""
# plugin/rpc
msgid "Failed to acquire a port between %d and %d"
msgstr ""
# plugin/screensaver
msgid "Screensaver"
msgstr ""

287
safeeyes/config/safeeyes.json
View File

@ -1,141 +1,152 @@
{
"meta": {
"config_version": "6.0.4"
"meta": {
"config_version": "6.0.4"
},
"random_order": true,
"short_break_interval": 15,
"long_break_interval": 75,
"long_break_duration": 60,
"pre_break_warning_time": 10,
"short_break_duration": 15,
"persist_state": false,
"postpone_duration": 5,
"short_breaks": [
{
"name": "Tightly close your eyes"
},
"random_order": true,
"short_break_interval": 15,
"long_break_interval": 75,
"long_break_duration": 60,
"pre_break_warning_time": 10,
"short_break_duration": 15,
"persist_state": false,
"postpone_duration": 5,
"use_rpc_server": true,
"rpc_port": 7200,
"short_breaks": [{
"name": "Tightly close your eyes"
},
{
"name": "Roll your eyes a few times to each side"
},
{
"name": "Rotate your eyes in clockwise direction"
},
{
"name": "Rotate your eyes in counterclockwise direction"
},
{
"name": "Blink your eyes"
},
{
"name": "Focus on a point in the far distance"
},
{
"name": "Have some water"
}
],
"long_breaks": [{
"name": "Walk for a while"
},
{
"name": "Lean back at your seat and relax"
}
],
"plugins": [{
"id": "donotdisturb",
"enabled": true,
"version": "0.0.2",
"settings": {
"skip_break_windows": "",
"take_break_windows": "",
"unfullscreen": true,
"while_on_battery": false
}
},
{
"id": "notification",
"enabled": true,
"version": "0.0.1"
},
{
"id": "audiblealert",
"enabled": true,
"version": "0.0.3",
"settings": {
"pre_break_alert": true,
"post_break_alert": true
}
},
{
"id": "trayicon",
"enabled": true,
"version": "0.0.3",
"settings": {
"show_time_in_tray": false,
"allow_disabling": true,
"disable_options": [{
"time": 30,
"unit": "minute"
},
{
"time": 1,
"unit": "hour"
},
{
"time": 2,
"unit": "hour"
},
{
"time": 3,
"unit": "hour"
}
]
}
},
{
"id": "smartpause",
"enabled": true,
"version": "0.0.3",
"settings": {
"idle_time": 5,
"interpret_idle_as_break": false,
"postpone_if_active": false
}
},
{
"id": "screensaver",
"enabled": true,
"version": "0.0.2",
"settings": {
"command": "",
"min_seconds": 3
}
},
{
"id": "healthstats",
"enabled": false,
"version": "0.0.2",
"settings": {
"statistics_reset_interval": 24
}
},
{
"id": "mediacontrol",
"enabled": true,
"version": "0.0.1"
},
{
"id": "breakscreen",
"enabled": true,
"version": "0.0.1",
"settings": {
"allow_skipping": true,
"allow_postponing": false,
"postpone_duration": 5,
"keyboard_disabled_period": 2,
"skip_keyboard_shortcut": 9,
"postpone_keyboard_shortcut": 65
}
}
]
{
"name": "Roll your eyes a few times to each side"
},
{
"name": "Rotate your eyes in clockwise direction"
},
{
"name": "Rotate your eyes in counterclockwise direction"
},
{
"name": "Blink your eyes"
},
{
"name": "Focus on a point in the far distance"
},
{
"name": "Have some water"
}
],
"long_breaks": [
{
"name": "Walk for a while"
},
{
"name": "Lean back at your seat and relax"
}
],
"plugins": [
{
"id": "rpcserver",
"enabled": true,
"version": "0.0.1",
"settings": {
"port": 7200,
"max_attempts": 5
}
},
{
"id": "donotdisturb",
"enabled": true,
"version": "0.0.2",
"settings": {
"skip_break_windows": "",
"take_break_windows": "",
"unfullscreen": true,
"while_on_battery": false
}
},
{
"id": "notification",
"enabled": true,
"version": "0.0.1"
},
{
"id": "audiblealert",
"enabled": true,
"version": "0.0.3",
"settings": {
"pre_break_alert": true,
"post_break_alert": true
}
},
{
"id": "trayicon",
"enabled": true,
"version": "0.0.3",
"settings": {
"show_time_in_tray": false,
"allow_disabling": true,
"disable_options": [
{
"time": 30,
"unit": "minute"
},
{
"time": 1,
"unit": "hour"
},
{
"time": 2,
"unit": "hour"
},
{
"time": 3,
"unit": "hour"
}
]
}
},
{
"id": "smartpause",
"enabled": true,
"version": "0.0.3",
"settings": {
"idle_time": 5,
"interpret_idle_as_break": false,
"postpone_if_active": false
}
},
{
"id": "screensaver",
"enabled": true,
"version": "0.0.2",
"settings": {
"command": "",
"min_seconds": 3
}
},
{
"id": "healthstats",
"enabled": false,
"version": "0.0.2",
"settings": {
"statistics_reset_interval": 24
}
},
{
"id": "mediacontrol",
"enabled": true,
"version": "0.0.1"
},
{
"id": "breakscreen",
"enabled": true,
"version": "0.0.1",
"settings": {
"allow_skipping": true,
"allow_postponing": false,
"postpone_duration": 5,
"keyboard_disabled_period": 2,
"skip_keyboard_shortcut": 9,
"postpone_keyboard_shortcut": 65
}
}
]
}

4
safeeyes/context.py
View File

@ -16,7 +16,7 @@ from typing import Any
from safeeyes import SAFE_EYES_VERSION, utility
from safeeyes.config import Config
from safeeyes.env.desktop import DesktopEnvironment
from safeeyes.env.system import Environment
from safeeyes.spi.api import CoreAPI, BreakAPI, WindowAPI, PluginAPI, ThreadAPI
from safeeyes.spi.state import State
@ -59,7 +59,7 @@ class Context:
self.session: Session = Session(config.get('persist_state', False))
self.state = State.START
self.__settings_dialog_visible = False
self.env: DesktopEnvironment = DesktopEnvironment.get_env()
self.env: Environment = Environment()
self.core_api: CoreAPI
self.thread_api: ThreadAPI
self.break_api: BreakAPI

48
safeeyes/env/system.py vendored
View File

@ -17,11 +17,55 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import importlib.util
import os
import pwd
import shutil
import subprocess
from typing import List, Optional
import psutil
from safeeyes import SAFE_EYES_CONFIG_DIR, SAFE_EYES_HOME_DIR
from safeeyes.env.desktop import DesktopEnvironment
class Environment:
def __init__(self):
self.desktop: DesktopEnvironment = DesktopEnvironment.get_env()
self.user: str = Environment.__get_username()
self.found_another_safe_eyes = Environment.__is_another_safe_eyes_running()
@staticmethod
def __get_username():
return pwd.getpwuid(os.getuid())[0]
@staticmethod
def __is_another_safe_eyes_running() -> bool:
"""
Check if Safe Eyes is already running.
"""
process_count = 0
for proc in psutil.process_iter():
if not proc.cmdline:
continue
try:
# Check if safeeyes is in process arguments
if callable(proc.cmdline):
# Latest psutil has cmdline function
cmd_line = proc.cmdline()
else:
# In older versions cmdline was a list object
cmd_line = proc.cmdline
if ("python3" in cmd_line[0] or "python" in cmd_line[0]) and (
"safeeyes" in cmd_line[1] or "safeeyes" in cmd_line):
process_count += 1
if process_count > 1:
return True
# Ignore if process does not exist or does not have command line args
except (IndexError, psutil.NoSuchProcess):
pass
return False
def execute(command: List[str]) -> None:
@ -56,9 +100,9 @@ def get_resource_path(resource_name: str) -> Optional[str]:
"""
if resource_name is None:
return None
resource_location = os.path.join(SAFE_EYES_CONFIG_DIR, 'resource', resource_name)
resource_location = os.path.join(SAFE_EYES_CONFIG_DIR, "resource", resource_name)
if not os.path.isfile(resource_location):
resource_location = os.path.join(SAFE_EYES_HOME_DIR, 'resource', resource_name)
resource_location = os.path.join(SAFE_EYES_HOME_DIR, "resource", resource_name)
if not os.path.isfile(resource_location):
# Resource not found
return None

80
safeeyes/glade/settings_dialog.glade
View File

@ -665,86 +665,6 @@
<property name="position">2</property>
</packing>
</child>
<child>
<object class="GtkInfoBar" id="warn_bar_rpc_server">
<property name="app_paintable">True</property>
<property name="can_focus">False</property>
<property name="no_show_all">True</property>
<property name="message_type">warning</property>
<property name="show_close_button">True</property>
<signal name="close" handler="on_warn_bar_rpc_server_close" swapped="no"/>
<signal name="response" handler="on_warn_bar_rpc_server_close" swapped="no"/>
<child internal-child="action_area">
<object class="GtkButtonBox">
<property name="can_focus">False</property>
</object>
<packing>
<property name="expand">False</property>
<property name="fill">False</property>
<property name="position">0</property>
</packing>
</child>
<child internal-child="content_area">
<object class="GtkBox">
<property name="can_focus">False</property>
</object>
<packing>
<property name="expand">False</property>
<property name="fill">False</property>
<property name="position">0</property>
</packing>
</child>
<child>
<placeholder/>
</child>
<style>
<class name="warn_bar_rpc_server"/>
</style>
</object>
<packing>
<property name="expand">True</property>
<property name="fill">True</property>
<property name="position">3</property>
</packing>
</child>
<child>
<object class="GtkBox" id="box10">
<property name="visible">True</property>
<property name="can_focus">False</property>
<property name="spacing">5</property>
<child>
<object class="GtkLabel" id="lbl_toggle_rpc_server">
<property name="visible">True</property>
<property name="can_focus">False</property>
<property name="label" translatable="yes">Use RPC server to receive runtime commands</property>
</object>
<packing>
<property name="expand">False</property>
<property name="fill">True</property>
<property name="position">0</property>
</packing>
</child>
<child>
<object class="GtkSwitch" id="switch_rpc_server">
<property name="visible">True</property>
<property name="can_focus">True</property>
<property name="halign">end</property>
<property name="valign">center</property>
</object>
<packing>
<property name="expand">False</property>
<property name="fill">False</property>
<property name="pack_type">end</property>
<property name="position">1</property>
</packing>
</child>
</object>
<packing>
<property name="expand">False</property>
<property name="fill">True</property>
<property name="position">4</property>
</packing>
</child>
</object>
</child>
</object>

8
safeeyes/plugin_utils/loader.py
View File

@ -103,7 +103,6 @@ class PluginLoader:
new_settings = dict(plugin.get('settings', {}))
new_settings['path'] = os.path.join(plugin_dir, plugin_id)
plugin_obj.update_settings(new_settings)
plugin_obj.enable()
else:
# This is the first time to load the plugin
# Check for dependencies
@ -116,10 +115,11 @@ class PluginLoader:
logging.info("Successfully loaded '%s' plugin from '%s'", plugin['id'], str(module.__file__))
new_settings = dict(plugin.get('settings', {}))
new_settings['path'] = os.path.join(plugin_dir, plugin_id)
plugin_obj = PluginProxy(plugin['id'], module, plugin_enabled, plugin_config, new_settings)
plugin_obj = PluginProxy(plugin['id'], module, False, plugin_config, new_settings)
self.__plugins[plugin['id']] = plugin_obj
if plugin_enabled:
plugin_obj.enable()
if plugin_enabled:
plugin_obj.enable()
@staticmethod
def load_plugins_config(context: Context, config: Config) -> List[dict]:

14
safeeyes/plugin_utils/manager.py
View File

@ -32,6 +32,14 @@ class PluginManager(PluginAPI):
for plugin in self.__plugins:
plugin.init(context, {})
def enable(self) -> None:
for plugin in self.__plugins:
plugin.enable()
def disable(self) -> None:
for plugin in self.__plugins:
plugin.disable()
def get_break_action(self, break_obj: Break) -> BreakAction:
"""
This function is called before on_pre_break and on_start_break.
@ -90,3 +98,9 @@ class PluginManager(PluginAPI):
next_long_break: Optional[datetime]) -> None:
for plugin in self.__plugins:
plugin.update_next_break(break_obj, next_short_break, next_long_break)
def get_plugin(self, plugin_id: str) -> Optional[PluginProxy]:
for plugin in self.__plugins:
if plugin.get_id() == plugin_id:
return plugin
return None

6
safeeyes/plugin_utils/plugin.py
View File

@ -31,6 +31,12 @@ class Plugin(abc.ABC):
"""
pass
def enable(self) -> None:
pass
def disable(self) -> None:
pass
def get_break_action(self, break_obj: Break) -> Optional[BreakAction]:
"""
Called just before on_pre_break and on_start_break.

10
safeeyes/plugin_utils/proxy.py
View File

@ -47,6 +47,9 @@ class PluginProxy(Plugin):
self.__config: dict = plugin_config
self.__settings = plugin_settings
def get_id(self) -> str:
return self.__id
def is_enabled(self) -> bool:
return self.__enabled
@ -177,6 +180,13 @@ class PluginProxy(Plugin):
logging.debug("Call update_next_break of the plugin '%s'", self.__id)
self.__plugin.update_next_break(break_obj, next_short_break, next_long_break)
def execute_if_exists(self, func_name: str, *args, **kwargs) -> Optional[Any]:
if self.__enabled and hasattr(self.__plugin, func_name):
logging.debug("Call %s of the plugin '%s'", func_name, self.__id)
function = getattr(self.__plugin, func_name)
return function(*args, **kwargs)
return None
class ValidatorProxy(Validator):

2
safeeyes/plugins/breakscreen/plugin.py
View File

@ -209,7 +209,7 @@ class BreakScreen:
# Set visual to apply css theme. It should be called before show method.
window.set_visual(window.get_screen().get_rgba_visual())
if self.__context.env.name == "kde":
if self.__context.env.desktop.name == "kde":
# Fix flickering screen in KDE by setting opacity to 1
window.set_opacity(0.9)

2
safeeyes/plugins/donotdisturb/plugin.py
View File

@ -51,7 +51,7 @@ class SystemState:
if self.__dnd_while_on_battery and SystemState.__is_on_battery():
logging.debug("Do Not Disturb: skipping the break as the system is on battery")
return True
return self.__is_wayland_full_screen() if self.__context.env.is_wayland() else self.__is_xorg_full_screen()
return self.__is_wayland_full_screen() if self.__context.env.desktop.is_wayland() else self.__is_xorg_full_screen()
def __is_wayland_full_screen(self) -> bool:
logging.debug('Do Not Disturb: searching for full-screen application in wayland')

2
safeeyes/plugins/donotdisturb/validator.py
View File

@ -22,7 +22,7 @@ from safeeyes.context import Context
def validate(ctx: Context, plugin_config: dict, plugin_settings: dict) -> Optional[str]:
command = "wlrctl" if ctx.env.is_wayland() else "xprop"
command = "wlrctl" if ctx.env.desktop.is_wayland() else "xprop"
if not utility.command_exist(command):
return _("Please install the command-line tool '%s'") % command
else:

32
safeeyes/plugins/rpcserver/config.json Normal file
View File

@ -0,0 +1,32 @@
{
"meta": {
"name": "RPC Server",
"description": "RPC server to receive command-line arguments",
"version": "0.0.1"
},
"dependencies": {
"python_modules": [],
"shell_commands": [],
"operating_systems": [],
"desktop_environments": [],
"resources": []
},
"settings": [
{
"id": "port",
"label": "Port",
"type": "INT",
"default": 7200,
"max": 65535,
"min": 1024
},
{
"id": "max_attempts",
"label": "Maximum adjacent ports to try",
"type": "INT",
"default": 5,
"max": 1024,
"min": 0
}
]
}

BIN
safeeyes/plugins/rpcserver/icon.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 173 B

335
safeeyes/plugins/rpcserver/plugin.py Normal file
View File

@ -0,0 +1,335 @@
# 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/>.
import logging
import socket
from contextlib import closing
from typing import Optional
from xmlrpc.client import ServerProxy
from xmlrpc.server import SimpleXMLRPCServer
from safeeyes.context import Context
from safeeyes.thread import worker
class RPCClient:
"""
An RPC client to communicate with the RPC server.
"""
def __init__(self, port: int):
self.__port: int = port
self.proxy = ServerProxy("http://localhost:%d/" % self.__port, allow_none=True)
def show_settings(self):
"""
Show the settings dialog.
"""
self.proxy.show_settings()
def show_about(self):
"""
Show the about dialog.
"""
self.proxy.show_about()
def enable_safe_eyes(self):
"""
Enable Safe Eyes.
"""
self.proxy.enable_safe_eyes()
def disable_safe_eyes(self):
"""
Disable Safe Eyes.
"""
self.proxy.disable_safe_eyes()
def take_break(self):
"""
Take a break now.
"""
self.proxy.take_break()
def get_status(self):
"""
Return the status of Safe Eyes
"""
return self.proxy.get_status()
def get_user(self):
"""
Return the get_user of Safe Eyes
"""
return self.proxy.get_user()
def quit_safe_eyes(self):
"""
Quit Safe Eyes.
"""
self.proxy.quit_safe_eyes()
class RPCServer:
"""
An asynchronous RPC server.
"""
def __init__(self, context: Context, config: dict):
self.__running: bool = True
self.__context: Context = context
self.__port: Optional[int] = RPCServer.get_available_port(context.env.user,
int(config.get("port", 7200)),
int(config.get("max_attempts", 5)))
self.__server: Optional[SimpleXMLRPCServer] = None
if self.__port is None:
logging.debug("RPC Server: Cannot create an RPC server")
else:
logging.debug("Init the rpc server to listen on port %d", self.__port)
self.__server = SimpleXMLRPCServer(("localhost", self.__port), logRequests=False, allow_none=True)
self.__server.timeout = 1.0
self.__server.register_function(self.show_settings)
self.__server.register_function(self.show_about)
self.__server.register_function(self.enable_safe_eyes)
self.__server.register_function(self.disable_safe_eyes)
self.__server.register_function(self.take_break)
self.__server.register_function(self.get_status)
self.__server.register_function(self.get_user)
self.__server.register_function(self.quit_safe_eyes)
def show_settings(self) -> None:
self.__context.window_api.show_settings()
def show_about(self) -> None:
self.__context.window_api.show_about()
def enable_safe_eyes(self) -> None:
self.__context.core_api.start()
def disable_safe_eyes(self) -> None:
self.__context.core_api.stop()
def quit_safe_eyes(self) -> None:
self.__context.core_api.quit()
def take_break(self) -> None:
self.__context.break_api.take_break()
def get_status(self) -> str:
return self.__context.core_api.get_status()
def get_user(self) -> str:
return self.__context.env.user
@worker
def start(self):
"""
Start the RPC server.
"""
try:
logging.debug("RPC Server: starting the server listening on port %d", self.__port)
while self.__server is not None and self.__running:
self.__server.handle_request()
self.__server.server_close()
logging.debug("RPC Server: stopped the server successfully")
finally:
self.__running = True
def stop(self):
"""
Stop the server.
"""
if self.__server is not None and self.__running:
logging.debug("RPC Server: stopping the server listening on port %d", self.__port)
self.__running = False
@staticmethod
def is_available(port: int) -> bool:
with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
return sock.connect_ex(("127.0.0.1", port)) != 0
@staticmethod
def get_available_port(user: str, port: int, max_attempts: int) -> Optional[int]:
i = 0
original_port = port
while i < max_attempts:
i += 1
if RPCServer.is_available(port):
return port
else:
rpc_client = RPCClient(port)
try:
if user == rpc_client.get_user():
# Safe Eyes is running under the same user
logging.debug("RPC Server: Another instance of Safe Eyes is running under the same user %s",
user)
return None
except ConnectionError:
# Port is used by some other process
pass
port += 1
logging.error(
"RPC Server: Failed to get a port between %d and %d. Change the port number of the RPC Server plugin.",
original_port, port)
return None
@staticmethod
def get_safe_eyes_port(user: str, port: int, max_attempts: int) -> Optional[int]:
i = 0
original_port = port
while i < max_attempts:
i += 1
rpc_client = RPCClient(port)
try:
if user == rpc_client.get_user():
# Safe Eyes is running under the same user
return port
except ConnectionError:
# Port is used by some other process
pass
port += 1
return None
server: Optional[RPCServer] = None
client: Optional[RPCClient] = None
def init(ctx: Context, plugin_config: dict) -> None:
"""
Create the server.
"""
global server
global client
if server:
server.stop()
if ctx.env.found_another_safe_eyes:
# Do not start the RPC server but create the client.
port = int(plugin_config.get("port", 7200))
max_attempts = int(plugin_config.get("max_attempts", 5))
safe_eyes_port = RPCServer.get_safe_eyes_port(ctx.env.user, port, max_attempts)
if safe_eyes_port:
logging.debug("RPC Server: Found another Safe Eyes running at %d", safe_eyes_port)
client = RPCClient(safe_eyes_port)
else:
client = None
server = None
else:
server = RPCServer(ctx, plugin_config)
server.start()
client = None
def enable() -> None:
if server:
server.start()
def disable() -> None:
if server:
server.stop()
def on_exit() -> None:
if server:
server.stop()
def show_settings() -> None:
"""
Show the settings dialog.
"""
if client is None:
return
try:
client.show_settings()
except ConnectionError:
logging.error("RPC Server: Failed to establish a connection with the existing RPC server")
def show_about() -> None:
"""
Show the about dialog.
"""
if client is None:
return
try:
client.show_about()
except ConnectionError:
logging.error("RPC Server: Failed to establish a connection with the existing RPC server")
def enable_safe_eyes() -> None:
"""
Enable Safe Eyes.
"""
if client is None:
return
try:
client.enable_safe_eyes()
except ConnectionError:
logging.error("RPC Server: Failed to establish a connection with the existing RPC server")
def disable_safe_eyes() -> None:
"""
Disable Safe Eyes.
"""
if client is None:
return
try:
client.disable_safe_eyes()
except ConnectionError:
logging.error("RPC Server: Failed to establish a connection with the existing RPC server")
def take_break() -> None:
"""
Take a break now.
"""
if client is None:
return
try:
client.take_break()
except ConnectionError:
logging.error("RPC Server: Failed to establish a connection with the existing RPC server")
def get_status() -> Optional[str]:
"""
Return the status of Safe Eyes
"""
if client is None:
return None
try:
return client.get_status()
except ConnectionError:
logging.error("RPC Server: Failed to establish a connection with the existing RPC server")
return None
def quit_safe_eyes() -> None:
"""
Quit Safe Eyes.
"""
if client is None:
return
try:
client.quit_safe_eyes()
except ConnectionError:
logging.error("RPC Server: Failed to establish a connection with the existing RPC server")

50
safeeyes/plugins/rpcserver/validator.py Normal file
View File

@ -0,0 +1,50 @@
# Safe Eyes is a utility to remind you to take break frequently
# to protect your eyes from eye strain.
# Copyright (C) 2021 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/>.
import socket
from contextlib import closing
from typing import Optional
from xmlrpc.client import ServerProxy
from safeeyes.context import Context
from safeeyes.util.locale import _
def is_available(port: int) -> bool:
with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
return sock.connect_ex(("127.0.0.1", port)) != 0
def validate(ctx: Context, plugin_config: dict, plugin_settings: dict) -> Optional[str]:
port = int(plugin_settings.get("port", 7200))
original_port = port
max_attempts = int(plugin_settings.get("max_attempts", 5))
i = 0
while i < max_attempts:
i += 1
if is_available(port):
return None
else:
try:
client = ServerProxy("http://localhost:%d/" % port, allow_none=True)
if ctx.env.user == client.get_user():
# Another Safe Eyes is running under the same user
return None
except BaseException:
# Port is used by some other process
port += 1
return _("Failed to acquire a port between %d and %d") % (original_port, port)

3
safeeyes/plugins/screensaver/plugin.py
View File

@ -19,7 +19,6 @@
Screensaver plugin locks the desktop using native screensaver application, after long breaks.
"""
import logging
import os
from typing import Optional, List
@ -40,7 +39,7 @@ class Screensaver:
self.min_seconds: int = config["min_seconds"]
self.__tray_icon_path: str = os.path.join(config["path"], "resource/lock.png")
self.__command: List[str] = config["command"].split() if config[
"command"] else Screensaver.__lock_screen_command(ctx.env.name)
"command"] else Screensaver.__lock_screen_command(ctx.env.desktop.name)
self.__lock_required = False
def reset(self) -> None:

6
safeeyes/plugins/smartpause/plugin.py
View File

@ -42,7 +42,7 @@ class SmartPause:
self.__short_break_interval = context.config.get("short_break_interval") * 60 # Convert to seconds
self.__long_break_duration = context.config.get("long_break_duration")
self.__waiting_time = min(2, self.__idle_time) # If idle time is 1 sec, wait only 1 sec
self.__is_wayland_and_gnome = context.env.name == "gnome" and context.env.is_wayland()
self.__is_wayland_and_gnome = context.env.desktop.name == "gnome" and context.env.desktop.is_wayland()
self.__active: bool = False
self.__smart_pause_activated: bool = False
self.__next_break_time: datetime.datetime = None
@ -52,7 +52,6 @@ class SmartPause:
def start(self) -> None:
if not self.__is_active():
# If SmartPause is already started, do not start it again
logging.debug("Start Smart Pause plugin")
self.__set_active(True)
self.__start_idle_monitor()
@ -119,7 +118,7 @@ class SmartPause:
if system_idle_time >= self.__idle_time and self.__context.state == State.WAITING:
self.__smart_pause_activated = True
self.__idle_start_time = datetime.datetime.now() - datetime.timedelta(seconds=system_idle_time)
logging.debug("Smart Pause: pause Safe Eyes due to system idle")
logging.debug("Smart Pause: pause Safe Eyes due to system idle for %d seconds", system_idle_time)
self.__context.core_api.stop()
elif system_idle_time < self.__idle_time and self.__context.state == State.STOPPED and self.__idle_start_time is not None:
@ -194,7 +193,6 @@ def init(context: Context, plugin_config: dict):
"""
Initialize the plugin.
"""
logging.info("Smart Pause: initialize the plugin")
global smart_pause
smart_pause = SmartPause(context, plugin_config)

2
safeeyes/plugins/smartpause/validator.py
View File

@ -23,7 +23,7 @@ from safeeyes.util.locale import _
def validate(ctx: Context, plugin_config: dict, plugin_settings: dict) -> Optional[str]:
command = "dbus-send" if ctx.env.name == "gnome" and ctx.env.is_wayland() else "xprintidle"
command = "dbus-send" if ctx.env.desktop.name == "gnome" and ctx.env.desktop.is_wayland() else "xprintidle"
if not utility.command_exist(command):
return _("Please install the command-line tool '%s'") % command
else:

4
safeeyes/plugins/trayicon/plugin.py
View File

@ -326,7 +326,7 @@ class TrayIcon:
Change the UI to disabled state.
"""
if self.__active:
logging.debug("Tray Icon: disable Safe Eyes")
logging.debug("Tray Icon: disable the icon")
self.__active = False
self.__indicator.set_icon("safeeyes_disabled")
self.__item_info.set_label(_("Disabled until restart"))
@ -342,7 +342,7 @@ class TrayIcon:
Change the UI to enabled state.
"""
if not self.__active:
logging.debug("Tray Icon: enable Safe Eyes")
logging.debug("Tray Icon: enable the icon")
self.__active = True
self.__indicator.set_icon("safeeyes_enabled")
self.__item_info.set_sensitive(True)

115
safeeyes/rpc.py
View File

@ -1,115 +0,0 @@
# 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/>.
"""
RPC server and client implementation.
"""
import logging
from threading import Thread
from xmlrpc.server import SimpleXMLRPCServer
from xmlrpc.client import ServerProxy
from safeeyes.context import Context
class RPCServer:
"""
An aynchronous RPC server.
"""
def __init__(self, context:Context):
self.__running = False
port = context.config.get('rpc_port')
logging.info('Setting up an RPC server on port %d', port)
# self.__server = SimpleXMLRPCServer(("localhost", port), logRequests=False, allow_none=True)
# self.__server.register_function(context['api']['show_settings'], 'show_settings')
# self.__server.register_function(context['api']['show_about'], 'show_about')
# self.__server.register_function(context['api']['enable_safeeyes'], 'enable_safeeyes')
# self.__server.register_function(context['api']['disable_safeeyes'], 'disable_safeeyes')
# self.__server.register_function(context['api']['take_break'], 'take_break')
# self.__server.register_function(context['api']['status'], 'status')
# self.__server.register_function(context['api']['quit'], 'quit')
def start(self):
"""
Start the RPC server.
"""
if not self.__running:
self.__running = True
logging.info('Start the RPC server')
server_thread = Thread(target=self.__server.serve_forever)
server_thread.start()
def stop(self):
"""
Stop the server.
"""
if self.__running:
logging.info('Stop the RPC server')
self.__running = False
self.__server.shutdown()
class RPCClient:
"""
An RPC client to communicate with the RPC server.
"""
def __init__(self, port):
self.port = port
self.proxy = ServerProxy('http://localhost:%d/' % self.port, allow_none=True)
def show_settings(self):
"""
Show the settings dialog.
"""
self.proxy.show_settings()
def show_about(self):
"""
Show the about dialog.
"""
self.proxy.show_about()
def enable_safeeyes(self):
"""
Enable Safe Eyes.
"""
self.proxy.enable_safeeyes()
def disable_safeeyes(self):
"""
Disable Safe Eyes.
"""
self.proxy.disable_safeeyes(None)
def take_break(self):
"""
Take a break now.
"""
self.proxy.take_break()
def status(self):
"""
Return the status of Safe Eyes
"""
return self.proxy.status()
def quit(self):
"""
Quit Safe Eyes.
"""
self.proxy.quit()

27
safeeyes/safeeyes.py
View File

@ -28,6 +28,7 @@ import dbus
import gi
from dbus.mainloop.glib import DBusGMainLoop
from safeeyes.args import Arguments
from safeeyes.breaks.scheduler import BreakScheduler
from safeeyes.config import Config
from safeeyes.context import Context
@ -36,8 +37,8 @@ from safeeyes.plugin_utils.manager import PluginManager
from safeeyes.spi.api import CoreAPI
from safeeyes.spi.state import State
from safeeyes.thread import Heartbeat, main
from safeeyes.ui.UIManager import UIManager
from safeeyes.util import locale
from safeeyes.ui.manager import UIManager
from safeeyes.util.locale import _
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
@ -48,8 +49,8 @@ class SafeEyes(CoreAPI):
This class represents a runnable Safe Eyes instance.
"""
def __init__(self, system_locale, config: Config):
self.__context: Context = Context(config, locale.init_locale())
def __init__(self, system_locale):
self.__context: Context = Context(Config(), system_locale)
self.__plugin_loader = PluginLoader()
self.__heartbeat = Heartbeat(self.__context)
self.__plugin_manager: PluginManager = PluginManager(self.__plugin_loader.load(self.__context))
@ -62,6 +63,22 @@ class SafeEyes(CoreAPI):
# Save the session on exit
atexit.register(self.__persist_session)
def run(self, args):
arguments = Arguments(self.__context, self.__plugin_manager, args)
if self.__context.env.found_another_safe_eyes:
logging.info("Safe Eyes is already running")
arguments.execute_on_remote_instance()
sys.exit(0)
else:
if args.status:
print(_('Safe Eyes is not running'))
sys.exit(0)
elif not args.quit:
self.start()
arguments.execute_on_local_instance()
Gtk.main()
def start(self, scheduled_next_break_time: datetime.datetime = None, reset_breaks=False):
"""
Listen to tray icon enable action and send the signal to core.
@ -99,6 +116,8 @@ class SafeEyes(CoreAPI):
def quit(self):
self.stop()
logging.info("Quit safe eyes")
self.__plugin_manager.disable()
self.__plugin_manager.on_exit()
self.__context.state = State.QUIT
Gtk.main_quit()
# Exit all threads

0
safeeyes/ui/UIManager.py → safeeyes/ui/manager.py
View File

30
safeeyes/ui/settings_dialog.py
View File

@ -56,7 +56,6 @@ class SettingsDialog:
self.last_short_break_interval = config.get('short_break_interval')
self.initializing = True
self.infobar_long_break_shown = False
self.warn_bar_rpc_server_shown = False
builder = utility.create_gtk_builder(SETTINGS_DIALOG_GLADE)
builder.connect_signals(self)
@ -74,21 +73,11 @@ class SettingsDialog:
self.spin_time_to_prepare = builder.get_object('spin_time_to_prepare')
self.switch_random_order = builder.get_object('switch_random_order')
self.switch_persist = builder.get_object('switch_persist')
self.switch_rpc_server = builder.get_object('switch_rpc_server')
self.info_bar_long_break = builder.get_object("info_bar_long_break")
self.warn_bar_rpc_server = builder.get_object("warn_bar_rpc_server")
self.info_bar_long_break.hide()
self.warn_bar_rpc_server.hide()
# Set the current values of input fields
self.__initialize(config)
# Update relative states
# GtkSwitch state-set signal is available only from 3.14
if Gtk.get_minor_version() >= 14:
# Add event listener to RPC server switch
self.switch_rpc_server.connect('state-set', self.on_switch_rpc_server_activate)
self.on_switch_rpc_server_activate(self.switch_rpc_server, self.switch_rpc_server.get_active())
self.initializing = False
def __initialize(self, config):
@ -109,7 +98,6 @@ class SettingsDialog:
self.spin_time_to_prepare.set_value(config.get('pre_break_warning_time'))
self.switch_random_order.set_active(config.get('random_order'))
self.switch_persist.set_active(config.get('persist_state'))
self.switch_rpc_server.set_active(config.get('use_rpc_server'))
self.infobar_long_break_shown = False
def __create_break_item(self, break_config, is_short):
@ -283,23 +271,6 @@ class SettingsDialog:
"""
self.info_bar_long_break.hide()
def on_switch_rpc_server_activate(self, switch, enabled):
"""
Event handler to the state change of the rpc server switch.
Show or hide the self.warn_bar_rpc_server based on the state of the rpc server.
"""
if not self.initializing and not enabled and not self.warn_bar_rpc_server_shown:
self.warn_bar_rpc_server_shown = True
self.warn_bar_rpc_server.show()
if enabled:
self.warn_bar_rpc_server.hide()
def on_warn_bar_rpc_server_close(self, warnbar, *user_data):
"""
Event handler for warning bar close action.
"""
self.warn_bar_rpc_server.hide()
def add_break(self, button):
"""
Event handler for add break button.
@ -319,7 +290,6 @@ class SettingsDialog:
self.__config.set('pre_break_warning_time', self.spin_time_to_prepare.get_value_as_int())
self.__config.set('random_order', self.switch_random_order.get_active())
self.__config.set('persist_state', self.switch_persist.get_active())
self.__config.set('use_rpc_server', self.switch_rpc_server.get_active())
for plugin in self.__config.get('plugins'):
if plugin['id'] in self.plugin_switches:
plugin['enabled'] = self.plugin_switches[plugin['id']].get_active()

7
safeeyes/util/locale.py
View File

@ -1,5 +1,6 @@
import gettext
import locale
import logging
from safeeyes import utility
@ -13,4 +14,8 @@ def init_locale():
def _(message: str) -> str:
return gettext.gettext(message)
try:
return gettext.gettext(message)
except BaseException:
logging.exception("Error in getting the translation for %s", message)
return message