Merge pull request slgobinath#747 from deltragon/typing-context

typing: add Context class for gradual typing

Author: deltragon

safeeyes/context.py

@@ -0,0 +1,137 @@
+# Safe Eyes is a utility to remind you to take break frequently
+# to protect your eyes from eye strain.
+
+# Copyright (C) 2025 Mel Dafert <[email protected]>
+
+# 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 collections.abc import MutableMapping
+import datetime
+import typing
+
+from safeeyes import utility
+from safeeyes.model import State
+
+if typing.TYPE_CHECKING:
+    from safeeyes.safeeyes import SafeEyes
+
+
+class API:
+    _application: "SafeEyes"
+
+    def __init__(
+        self,
+        application: "SafeEyes",
+    ) -> None:
+        self._application = application
+
+    def __getitem__(self, key: str) -> typing.Callable:
+        """This is soft-deprecated - it is preferred to access the property."""
+        return getattr(self, key)
+
+    def show_settings(self) -> None:
+        utility.execute_main_thread(self._application.show_settings)
+
+    def show_about(self) -> None:
+        utility.execute_main_thread(self._application.show_about)
+
+    def enable_safeeyes(self, next_break_time=-1) -> None:
+        utility.execute_main_thread(self._application.enable_safeeyes, next_break_time)
+
+    def disable_safeeyes(self, status=None, is_resting=False) -> None:
+        utility.execute_main_thread(
+            self._application.disable_safeeyes, status, is_resting
+        )
+
+    def status(self) -> str:
+        return self._application.status()
+
+    def quit(self) -> None:
+        utility.execute_main_thread(self._application.quit)
+
+    def take_break(self, break_type=None) -> None:
+        self._application.take_break(break_type)
+
+    def has_breaks(self, break_type=None) -> bool:
+        return self._application.safe_eyes_core.has_breaks(break_type)
+
+    def postpone(self, duration=-1) -> None:
+        self._application.safe_eyes_core.postpone(duration)
+
+    def get_break_time(self, break_type=None) -> typing.Optional[datetime.datetime]:
+        return self._application.safe_eyes_core.get_break_time(break_type)
+
+
+class Context(MutableMapping):
+    version: str
+    api: API
+    desktop: str
+    is_wayland: bool
+    locale: str
+    session: dict[str, typing.Any]
+    state: State
+
+    skipped: bool = False
+    postponed: bool = False
+    skip_button_disabled: bool = False
+    postpone_button_disabled: bool = False
+
+    ext: dict
+
+    def __init__(
+        self,
+        api: API,
+        locale: str,
+        version: str,
+        session: dict[str, typing.Any],
+    ) -> None:
+        self.version = version
+        self.desktop = utility.desktop_environment()
+        self.is_wayland = utility.is_wayland()
+        self.locale = locale
+        self.session = session
+        self.state = State.START
+        self.api = api
+
+        self.ext = {}
+
+    def __setitem__(self, key: str, value: typing.Any) -> None:
+        """This is soft-deprecated - it is preferred to access the property."""
+        if hasattr(self, key):
+            setattr(self, key, value)
+            return
+
+        self.ext[key] = value
+
+    def __getitem__(self, key: str) -> typing.Any:
+        """This is soft-deprecated - it is preferred to access the property."""
+        if hasattr(self, key):
+            return getattr(self, key)
+
+        return self.ext[key]
+
+    def __delitem__(self, key: str) -> None:
+        """This is soft-deprecated - it is preferred to access the property."""
+        if hasattr(self, key):
+            raise Exception("cannot delete property")
+
+        del self.ext[key]
+
+    def __len__(self) -> int:
+        """This is soft-deprecated."""
+        return len(self.ext)
+
+    def __iter__(self) -> typing.Iterator[typing.Any]:
+        """This is soft-deprecated."""
+        return iter(self.ext)

safeeyes/core.py

@@ -29,6 +29,8 @@
 from safeeyes.model import State
 from safeeyes.model import Config
 
+from safeeyes.context import Context
+
 import gi
 
 gi.require_version("GLib", "2.0")
@@ -45,6 +47,7 @@ class SafeEyesCore:
     postpone_duration: int = 0
     default_postpone_duration: int = 0
     pre_break_warning_time: int = 0
+    context: Context
 
     _break_queue: typing.Optional[BreakQueue] = None
 
@@ -62,7 +65,7 @@ class SafeEyesCore:
     # set to true when a break was requested
     _take_break_now: bool = False
 
-    def __init__(self, context) -> None:
+    def __init__(self, context: Context) -> None:
         """Create an instance of SafeEyesCore and initialize the variables."""
         # This event is fired before <time-to-prepare> for a break
         self.on_pre_break = EventHook()
@@ -77,11 +80,7 @@ def __init__(self, context) -> None:
         # This event is fired when deciding the next break time
         self.on_update_next_break = EventHook()
         self.context = context
-        self.context["skipped"] = False
-        self.context["postponed"] = False
-        self.context["skip_button_disabled"] = False
-        self.context["postpone_button_disabled"] = False
-        self.context["state"] = State.WAITING
+        self.context.state = State.WAITING
 
     def initialize(self, config: Config):
         """Initialize the internal properties from configuration."""
@@ -116,14 +115,14 @@ def stop(self, is_resting=False) -> None:
         self.paused_time = datetime.datetime.now().timestamp()
         # Stop the break thread
         self.running = False
-        if self.context["state"] != State.QUIT:
-            self.context["state"] = State.RESTING if (is_resting) else State.STOPPED
+        if self.context.state != State.QUIT:
+            self.context.state = State.RESTING if (is_resting) else State.STOPPED
 
         self.__wakeup_scheduler()
 
     def skip(self) -> None:
         """User skipped the break using Skip button."""
-        self.context["skipped"] = True
+        self.context.skipped = True
 
     def postpone(self, duration=-1) -> None:
         """User postponed the break using Postpone button."""
@@ -132,7 +131,7 @@ def postpone(self, duration=-1) -> None:
         else:
             self.postpone_duration = self.default_postpone_duration
         logging.debug("Postpone the break for %d seconds", self.postpone_duration)
-        self.context["postponed"] = True
+        self.context.postponed = True
 
     def get_break_time(
         self, break_type: typing.Optional[BreakType] = None
@@ -154,7 +153,7 @@ def take_break(self, break_type: typing.Optional[BreakType] = None) -> None:
         """
         if self._break_queue is None:
             return
-        if not self.context["state"] == State.WAITING:
+        if not self.context.state == State.WAITING:
             return
 
         if break_type is not None and self._break_queue.get_break().type != break_type:
@@ -189,7 +188,7 @@ def __scheduler_job(self) -> None:
         current_time = datetime.datetime.now()
         current_timestamp = current_time.timestamp()
 
-        if self.context["state"] == State.RESTING and self.paused_time > -1:
+        if self.context.state == State.RESTING and self.paused_time > -1:
             # Safe Eyes was resting
             paused_duration = int(current_timestamp - self.paused_time)
             self.paused_time = -1
@@ -203,11 +202,11 @@ def __scheduler_job(self) -> None:
                 # Skip the next long break
                 self._break_queue.skip_long_break()
 
-        if self.context["postponed"]:
+        if self.context.postponed:
             # Previous break was postponed
             logging.info("Prepare for postponed break")
             time_to_wait = self.postpone_duration
-            self.context["postponed"] = False
+            self.context.postponed = False
         elif current_timestamp < self.scheduled_next_break_timestamp:
             # Non-standard break was set.
             time_to_wait = round(
@@ -221,7 +220,7 @@ def __scheduler_job(self) -> None:
         self.scheduled_next_break_time = current_time + datetime.timedelta(
             seconds=time_to_wait
         )
-        self.context["state"] = State.WAITING
+        self.context.state = State.WAITING
         self.__fire_on_update_next_break(self.scheduled_next_break_time)
 
         # Wait for the pre break warning period
@@ -262,7 +261,7 @@ def __fire_pre_break(self) -> None:
         if self._break_queue is None:
             # This will only be called by methods which check this
             return
-        self.context["state"] = State.PRE_BREAK
+        self.context.state = State.PRE_BREAK
         proceed = self.__fire_hook(self.on_pre_break, self._break_queue.get_break())
         if not proceed:
             # Plugins wanted to ignore this break
@@ -298,9 +297,9 @@ def __do_start_break(self) -> None:
             # Plugins want to ignore this break
             self.__start_next_break()
             return
-        if self.context["postponed"]:
+        if self.context.postponed:
             # Plugins want to postpone this break
-            self.context["postponed"] = False
+            self.context.postponed = False
 
             if self.scheduled_next_break_time is None:
                 raise Exception("this should never happen")
@@ -322,7 +321,7 @@ def __start_break(self) -> None:
         if self._break_queue is None:
             # This will only be called by methods which check this
             return
-        self.context["state"] = State.BREAK
+        self.context.state = State.BREAK
         break_obj = self._break_queue.get_break()
         self._taking_break = break_obj
         self._countdown = break_obj.duration
@@ -340,8 +339,8 @@ def __cycle_break_countdown(self) -> None:
         if (
             self._countdown > 0
             and self.running
-            and not self.context["skipped"]
-            and not self.context["postponed"]
+            and not self.context.skipped
+            and not self.context.postponed
         ):
             countdown = self._countdown
             self._countdown -= 1
@@ -359,14 +358,14 @@ def __cycle_break_countdown(self) -> None:
 
     def __fire_stop_break(self) -> None:
         # Loop terminated because of timeout (not skipped) -> Close the break alert
-        if not self.context["skipped"] and not self.context["postponed"]:
+        if not self.context.skipped and not self.context.postponed:
             logging.info("Break is terminated automatically")
             self.__fire_hook(self.on_stop_break)
 
         # Reset the skipped flag
-        self.context["skipped"] = False
-        self.context["skip_button_disabled"] = False
-        self.context["postpone_button_disabled"] = False
+        self.context.skipped = False
+        self.context.skip_button_disabled = False
+        self.context.postpone_button_disabled = False
         self.__start_next_break()
 
     def __wait_for(
@@ -440,7 +439,7 @@ def __start_next_break(self) -> None:
         if self._break_queue is None:
             # This will only be called by methods which check this
             return
-        if not self.context["postponed"]:
+        if not self.context.postponed:
             self._break_queue.next()
 
         if self.running:

safeeyes/model.py

@@ -38,6 +38,9 @@
 from safeeyes import utility
 from safeeyes.translations import translate as _
 
+if typing.TYPE_CHECKING:
+    from safeeyes.context import Context
+
 
 class BreakType(Enum):
     """Type of Safe Eyes breaks."""
@@ -105,9 +108,12 @@ class BreakQueue:
     __is_random_order: bool
     __long_queue: typing.Optional[list[Break]]
     __short_queue: typing.Optional[list[Break]]
+    context: "Context"
 
     @classmethod
-    def create(cls, config: "Config", context) -> typing.Optional["BreakQueue"]:
+    def create(
+        cls, config: "Config", context: "Context"
+    ) -> typing.Optional["BreakQueue"]:
         short_break_time = config.get("short_break_interval")
         long_break_time = config.get("long_break_interval")
         is_random_order = config.get("random_order")
@@ -142,7 +148,7 @@ def create(cls, config: "Config", context) -> typing.Optional["BreakQueue"]:
 
     def __init__(
         self,
-        context,
+        context: "Context",
         short_break_time: int,
         long_break_time: int,
         is_random_order: bool,
@@ -166,7 +172,7 @@ def __init__(
         self.__set_next_break()
 
         # Restore the last break from session
-        last_break = context["session"].get("break")
+        last_break = context.session.get("break")
         if last_break is not None:
             current_break = self.get_break()
             if last_break != current_break.name:
@@ -247,7 +253,7 @@ def __set_next_break(self, break_type: typing.Optional[BreakType] = None) -> Non
             break_obj = self.__next_short()
 
         self.__current_break = break_obj
-        self.context["session"]["break"] = self.__current_break.name
+        self.context.session["break"] = self.__current_break.name
 
     def skip_long_break(self) -> None:
         if not (self.__short_queue and self.__long_queue):
@@ -265,7 +271,7 @@ def skip_long_break(self) -> None:
             # we could decrement the __current_long counter, but then we'd need to
             # handle wraparound and possibly randomizing, which seems complicated
             self.__current_break = self.__next_short()
-            self.context["session"]["break"] = self.__current_break.name
+            self.context.session["break"] = self.__current_break.name
 
     def is_empty(self, break_type: BreakType) -> bool:
         """Check if the given break type is empty or not."""
@@ -283,7 +289,7 @@ def __next_short(self) -> Break:
             raise Exception("this may only be called when there are short breaks")
 
         break_obj = shorts[self.__current_short]
-        self.context["break_type"] = "short"
+        self.context.ext["break_type"] = "short"
 
         # Update the index to next
         self.__current_short = (self.__current_short + 1) % len(shorts)
@@ -297,7 +303,7 @@ def __next_long(self) -> Break:
             raise Exception("this may only be called when there are long breaks")
 
         break_obj = longs[self.__current_long]
-        self.context["break_type"] = "long"
+        self.context.ext["break_type"] = "long"
 
         # Update the index to next
         self.__current_long = (self.__current_long + 1) % len(longs)