sankalp/Modus
1
0 Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat(config): switch configuration parsing to TOML and add google_lens support

Browse Source
This commit is contained in:
sankalp committed 2026-07-04 06:29:38 +05:45
1 parent bca67e255d
commit 89eb8db97b
5 files changed
+266 -80
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/services/config.py
+37 -10
View File
@@ -9,10 +9,13 @@ module that needs dynamic configuration.
from __future__ import annotations
import json
from typing import Any, Callable, Dict, List, Optional
from fabric.utils import Gio, GLib, get_relative_path, logger, os
from tomlkit import document as toml_document
from tomlkit import dump as toml_dump
from tomlkit import load as toml_load
from tomlkit.items import Integer, String, Bool, Array
class ConfigService:
@@ -30,11 +33,11 @@ class ConfigService:
return
self._initialized = True
self._config: Dict[str, Any] = {}
self._config: Any = {}
self._reload_callbacks: List[
Callable[[Dict[str, Any], Dict[str, Any]], None]
] = []
self._config_file: str = get_relative_path("../../config/config.json")
self._config_file: str = get_relative_path("../../config/config.toml")
self._monitors: List[Gio.FileMonitor] = []
self._reload_pending: bool = False
self.RELOAD_DELAY_MS: int = 100
@@ -61,9 +64,8 @@ class ConfigService:
try:
os.makedirs(os.path.dirname(self._config_file), exist_ok=True)
with open(self._config_file, "w") as f:
json.dump(self._config, f, indent=4)
toml_dump(self._config, f)
# Manually trigger a reload to notify listeners immediately
GLib.idle_add(self._reload_config)
return True
except Exception as e:
@@ -90,21 +92,46 @@ class ConfigService:
logger.error(f"An error occurred: {e}")
self._monitors.clear()
@staticmethod
def _dict_to_toml(d: dict) -> Any:
doc = toml_document()
for k, v in d.items():
if isinstance(v, bool):
doc[k] = Bool(v)
elif isinstance(v, int):
doc[k] = Integer(v)
elif isinstance(v, str):
doc[k] = String(v, String.STANDARD)
elif isinstance(v, list):
arr = Array()
for item in v:
if isinstance(item, bool):
arr.append(Bool(item))
elif isinstance(item, int):
arr.append(Integer(item))
elif isinstance(item, str):
arr.append(String(item, String.STANDARD))
else:
arr.append(item)
arr.set_multiline(True) if len(v) > 1 else None
doc[k] = arr
else:
doc[k] = v
return doc
def _load_config(self) -> None:
try:
if os.path.exists(self._config_file):
with open(self._config_file, "r") as f:
self._config = json.load(f)
self._config = toml_load(f)
else:
try:
from utils.constants import DEFAULT
# Ensure the config directory exists
os.makedirs(os.path.dirname(self._config_file), exist_ok=True)
self._config = self._dict_to_toml(DEFAULT)
with open(self._config_file, "w") as f:
json.dump(DEFAULT, f, indent=4)
self._config = DEFAULT.copy()
toml_dump(self._config, f)
logger.info(
f"[ConfigService] Generated default config at {self._config_file}"
)
src/services/google_lens.py
+176
View File
@@ -0,0 +1,176 @@
import contextlib
import tempfile
import webbrowser
from urllib.parse import quote
from fabric.core.service import Service
from fabric.utils import logger, os
from utils.functions import run_command
DEFAULT_UPLOADER = "https://litterbox.catbox.moe/resources/internals/api.php"
class GoogleLens(Service):
"""Service to capture region and search with Google Lens."""
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
@staticmethod
def get_initial():
if GoogleLens._instance is None:
GoogleLens._instance = GoogleLens()
return GoogleLens._instance
def __init__(self, **kwargs):
super().__init__(**kwargs)
def capture_region(self, dest_path):
# Check for dependencies
if run_command(["which", "grim"]).returncode != 0:
raise RuntimeError("grim not found - please install it")
if run_command(["which", "slurp"]).returncode != 0:
raise RuntimeError("slurp not found - please install it")
try:
# Select region
result = run_command(["slurp"], timeout=60)
if result.returncode != 0:
if result.returncode == 1:
raise RuntimeError("Selection cancelled")
raise RuntimeError(f"Failed to select region: {result.stderr}")
geom = (
result.stdout.strip()
if isinstance(result.stdout, str)
else result.stdout.decode("utf-8").strip()
)
if not geom:
raise RuntimeError("No region selected")
# Capture selected region
capture_result = run_command(["grim", "-g", geom, dest_path], timeout=60)
if capture_result.returncode != 0:
raise RuntimeError(f"Failed to capture region: {capture_result.stderr}")
except Exception as e:
raise RuntimeError(f"Failed to capture region: {e}")
def upload_file(self, uploader_url, file_path):
url_resp = None
try:
import mimetypes
import uuid
from urllib.request import Request, urlopen
boundary = f"----WebKitFormBoundary{uuid.uuid4().hex}"
headers = {"Content-Type": f"multipart/form-data; boundary={boundary}"}
data = []
fields = {
"reqtype": "fileupload",
"time": "1h",
}
for name, value in fields.items():
data.append(f"--{boundary}".encode())
data.append(f'Content-Disposition: form-data; name="{name}"'.encode())
data.append(b"")
data.append(value.encode())
filename = os.path.basename(file_path)
mime_type = mimetypes.guess_type(file_path)[0] or "application/octet-stream"
data.append(f"--{boundary}".encode())
data.append(
f'Content-Disposition: form-data; name="fileToUpload"; filename="{filename}"'.encode()
)
data.append(f"Content-Type: {mime_type}".encode())
data.append(b"")
with open(file_path, "rb") as f:
data.append(f.read())
data.append(f"--{boundary}--".encode())
data.append(b"")
body = b"\r\n".join(data)
req = Request(uploader_url, data=body, headers=headers)
with urlopen(req, timeout=60) as response:
url_resp = response.read().decode("utf-8").strip()
except Exception as e:
logger.error(f"Upload with urllib failed: {e}")
# Fallback to curl
if url_resp is None:
if run_command(["which", "curl"]).returncode != 0:
raise RuntimeError("Neither requests module nor curl available")
cmd = [
"curl",
"-sS",
"-F",
"reqtype=fileupload",
"-F",
"time=1h",
"-F",
f"fileToUpload=@{file_path}",
uploader_url,
]
result = run_command(cmd)
if result.returncode != 0:
raise RuntimeError(f"Upload failed: {result.stderr}")
url_resp = (
result.stdout.strip()
if isinstance(result.stdout, str)
else result.stdout.decode("utf-8").strip()
)
# catbox.moe returns the URL directly
if (
not url_resp
or not isinstance(url_resp, str)
or not url_resp.startswith("http")
):
raise RuntimeError(f"Invalid response from uploader: {url_resp}")
return url_resp
def search(self):
tmp = tempfile.NamedTemporaryFile(
prefix="google_lens_", suffix=".png", delete=False
)
tmp_path = tmp.name
tmp.close()
try:
# Capture region
logger.info("Select a region...")
self.capture_region(tmp_path)
# Upload
logger.info("Uploading...")
uploaded_url = self.upload_file(DEFAULT_UPLOADER, tmp_path)
logger.info(f"Uploaded: {uploaded_url}")
# Open in Lens
lens_url = f"https://lens.google.com/uploadbyurl?url={quote(uploaded_url, safe='')}"
logger.info("Opening Google Lens...")
webbrowser.open(lens_url)
except Exception as e:
logger.error(f"Error: {e}")
finally:
# Cleanup
if os.path.exists(tmp_path):
with contextlib.suppress(OSError):
os.unlink(tmp_path)
src/shared/data.py
+47 -67
View File
@@ -1,5 +1,4 @@
from fabric.utils import logger
import json
from fabric.utils import Gdk, GLib, get_relative_path, os
@@ -22,12 +21,12 @@ CURRENT_HEIGHT = screen.get_height()
WALLPAPERS_DIR_DEFAULT = get_relative_path("../assets/wallpapers_example/")
CONFIG_FILE = get_relative_path("../../config/config.json")
CONFIG_FILE = get_relative_path("../../config/config.toml")
MATUGEN_STATE_FILE = os.path.join(CONFIG_DIR, "matugen")
def load_config():
"""Load the configuration from config.json"""
"""Load the configuration from config.toml"""
try:
from services.config import start_config_service
@@ -39,79 +38,60 @@ def load_config():
if os.path.exists(CONFIG_FILE):
try:
import tomlkit
with open(CONFIG_FILE, "r") as f:
config = json.load(f)
config = dict(tomlkit.load(f))
except Exception as e:
logger.error(f"Error loading config: {e}")
return config
if os.path.exists(CONFIG_FILE):
with open(CONFIG_FILE, "r") as f:
config = json.load(f)
wallpapers_dir_from_config = config.get("wallpapers_dir", WALLPAPERS_DIR_DEFAULT)
WALLPAPERS_DIR = os.path.expanduser(wallpapers_dir_from_config)
DOCK_POSITION = config.get("dock_position", "Bottom")
DOCK_ENABLED = config.get("dock_enabled", True)
DOCK_AUTO_HIDE = config.get("dock_auto_hide", True)
DOCK_ALWAYS_OCCLUDED = config.get("dock_always_occluded", False)
DOCK_ICON_SIZE = config.get("dock_icon_size", 60)
WINDOW_SWITCHER_ITEMS_PER_ROW = config.get("window_switcher_items_per_row", 10)
HIDE_SPECIAL_WORKSPACE = config.get("hide_special_workspace", True)
DOCK_HIDE_SPECIAL_WORKSPACE_APPS = config.get(
"dock_hide_special_workspace_apps", True
)
def _load_config_from_toml():
"""Load config dict from config.toml (module-level helper)."""
if os.path.exists(CONFIG_FILE):
try:
import tomlkit
NOTIFICATION_TIMEOUT_STR = config.get("notification_timeout", "5s")
NOTIFICATION_TIMEOUT = parse_timeout_string(NOTIFICATION_TIMEOUT_STR)
NOTIFICATION_IGNORED_APPS_HISTORY = config.get(
"notification_ignored_apps_history", ["Hyprshot"]
)
NOTIFICATION_LIMITED_APPS_HISTORY = config.get(
"notification_limited_apps_history", ["Spotify"]
)
with open(CONFIG_FILE, "r") as f:
return dict(tomlkit.load(f))
except Exception as e:
logger.error(f"Error loading config: {e}")
return {}
PANEL_COMPONENTS_VISIBILITY = {
"imac_button": config.get("imac_button", True),
"systray": config.get("systray", True),
"control_center": config.get("control_center", True),
"search": config.get("search", True),
"global_menu": config.get("global_menu", True),
"network": config.get("network", True),
"battery": config.get("battery", True),
"notification_center": config.get("notification_center", True),
"workspace_indicator": config.get("workspace_indicator", True),
"bluetooth": config.get("bluetooth", True),
"date_time": config.get("date_time", True),
}
else:
WALLPAPERS_DIR = WALLPAPERS_DIR_DEFAULT
DOCK_POSITION = "Bottom"
DOCK_ENABLED = True
DOCK_ALWAYS_OCCLUDED = False
DOCK_AUTO_HIDE = True
DOCK_ICON_SIZE = 52
WINDOW_SWITCHER_ITEMS_PER_ROW = 10
HIDE_SPECIAL_WORKSPACE = True
DOCK_HIDE_SPECIAL_WORKSPACE_APPS = True
config = _load_config_from_toml()
wallpapers_dir_from_config = config.get("wallpapers_dir", WALLPAPERS_DIR_DEFAULT)
WALLPAPERS_DIR = os.path.expanduser(wallpapers_dir_from_config)
DOCK_POSITION = config.get("dock_position", "Bottom")
DOCK_ENABLED = config.get("dock_enabled", True)
DOCK_AUTO_HIDE = config.get("dock_auto_hide", True)
DOCK_ALWAYS_OCCLUDED = config.get("dock_always_occluded", False)
DOCK_ICON_SIZE = config.get("dock_icon_size", 60)
WINDOW_SWITCHER_ITEMS_PER_ROW = config.get("window_switcher_items_per_row", 10)
HIDE_SPECIAL_WORKSPACE = config.get("hide_special_workspace", True)
DOCK_HIDE_SPECIAL_WORKSPACE_APPS = config.get("dock_hide_special_workspace_apps", True)
NOTIFICATION_TIMEOUT_STR = "5s"
NOTIFICATION_TIMEOUT = parse_timeout_string(NOTIFICATION_TIMEOUT_STR)
NOTIFICATION_IGNORED_APPS_HISTORY = ["Hyprshot"]
NOTIFICATION_LIMITED_APPS_HISTORY = ["Spotify"]
NOTIFICATION_TIMEOUT_STR = config.get("notification_timeout", "5s")
NOTIFICATION_TIMEOUT = parse_timeout_string(NOTIFICATION_TIMEOUT_STR)
NOTIFICATION_IGNORED_APPS_HISTORY = config.get(
"notification_ignored_apps_history", ["Hyprshot"]
)
NOTIFICATION_LIMITED_APPS_HISTORY = config.get(
"notification_limited_apps_history", ["Spotify"]
)
PANEL_COMPONENTS_VISIBILITY = {
"imac_button": True,
"systray": True,
"control_center": True,
"search": True,
"global_menu": True,
"network": True,
"battery": True,
"notification_center": True,
"workspace_indicator": True,
"bluetooth": True,
"date_time": True,
}
PANEL_COMPONENTS_VISIBILITY = {
"imac_button": config.get("imac_button", True),
"systray": config.get("systray", True),
"control_center": config.get("control_center", True),
"search": config.get("search", True),
"global_menu": config.get("global_menu", True),
"network": config.get("network", True),
"battery": config.get("battery", True),
"notification_center": config.get("notification_center", True),
"workspace_indicator": config.get("workspace_indicator", True),
"bluetooth": config.get("bluetooth", True),
"date_time": config.get("date_time", True),
}
src/shared/window/applet_window.py
+4 -3
View File
@@ -11,7 +11,7 @@ class DismissLayer(Window):
self.event_box = EventBox()
super().__init__(
anchor="left right top bottom",
layer="top",
layer="overlay",
exclusivity="none",
keyboard_mode="none",
child=self.event_box,
@@ -70,10 +70,8 @@ class AppletWindow(PopupWindow):
self.dismiss_layer.show()
self.set_visible(True)
self.set_focus(None)
def close_applet(self):
if not self._is_open:
return
@@ -88,6 +86,9 @@ class AppletWindow(PopupWindow):
self.set_visible(False)
if hasattr(self, "_parent") and self._parent:
self._parent.set_visible(True)
if AppletWindow._active_popup == self:
AppletWindow._active_popup = None
src/utils/constants.py
+2
View File
@@ -21,6 +21,8 @@ DEFAULT = {
"workspace_indicator": True,
"bluetooth": True,
"date_time": True,
"google_lens": True,
"custom_mods": True,
"keyboard_layouts": ["us", "np"],
"window_switcher": True,
"osd": True,