1 Commits
Author SHA1 Message Date
Makussu 5b7398de2c feat: mail finder 2026-05-06 23:03:51 +02:00
33 changed files with 288 additions and 2305 deletions
-13
View File
@@ -1,7 +1,5 @@
height: 42
dev: true
buddy:
enable: true
window_title:
enable: true
vinyl:
@@ -18,17 +16,6 @@ notmuch:
screenrec:
enable: true
output_dir: "~/Videos/wl-screenrec"
org:
enable: true
paths:
- "~/Documents/notes"
todo_keywords: ["TODO", "NEXT", "IN-PROGRESS"]
done_keywords: ["DONE", "CANCELLED"]
counted_states: ["TODO", "NEXT", "IN-PROGRESS"]
update_interval: 60000
emacsclient_command: "emacsclient"
dropdown_width: 420
dropdown_height: 480
notifications:
enable: true
anchor: "top center"
+224 -332
View File
@@ -45,358 +45,250 @@
// {
homeManagerModules = {
sims =
{
config,
lib,
pkgs,
...
}:
let
cfg = config.services.sims;
{
config,
lib,
pkgs,
...
}:
let
cfg = config.services.sims;
settingsFormat = pkgs.formats.yaml { };
in
{
options.services.sims = {
enable = lib.mkEnableOption "sims status bar";
settingsFormat = pkgs.formats.yaml { };
in
{
options.services.sims = {
enable = lib.mkEnableOption "sims status bar";
package = lib.mkOption {
type = lib.types.package;
default = self.packages.${pkgs.system}.default;
description = "The sims package to use.";
};
package = lib.mkOption {
type = lib.types.package;
default = self.packages.${pkgs.system}.default;
description = "The sims package to use.";
};
settings = lib.mkOption {
type = lib.types.submodule {
options = {
vinyl = {
enable = lib.mkOption {
type = lib.types.bool;
default = false;
};
};
battery = {
enable = lib.mkOption {
type = lib.types.bool;
default = false;
};
};
buddy = {
enable = lib.mkOption {
type = lib.types.bool;
default = false;
description = "Whether to enable the bar buddy (animated pixel-art slime that reacts to system state)";
};
};
height = lib.mkOption {
type = lib.types.int;
default = 40;
description = "Height of the status bar in pixels";
};
logLevel = lib.mkOption {
type = lib.types.enum [
"TRACE"
"DEBUG"
"INFO"
"SUCCESS"
"WARNING"
"ERROR"
"CRITICAL"
];
default = "WARNING";
description = "Log level for the status bar (loguru levels: TRACE, DEBUG, INFO, SUCCESS, WARNING, ERROR, CRITICAL)";
};
window_title = {
enable = lib.mkOption {
type = lib.types.bool;
default = true;
description = "Whether to show the window title in the center of the bar";
};
};
stylix = lib.mkOption {
type = lib.types.attrsOf lib.types.anything;
default = {
enable = false;
};
description = "Stylix configuration passed from the stylix module";
};
calendar = {
enable = lib.mkOption {
type = lib.types.bool;
default = true;
description = "Whether to enable the calendar widget";
};
khal_path = lib.mkOption {
type = lib.types.str;
default = "khal";
description = "Path to the khal binary";
};
};
notmuch = {
enable = lib.mkOption {
type = lib.types.bool;
default = true;
description = "Whether to enable the notmuch email widget";
};
notmuch_path = lib.mkOption {
type = lib.types.str;
default = "notmuch";
description = "Path to the notmuch binary";
};
emacsclient_command = lib.mkOption {
type = lib.types.str;
default = "emacsclient";
description = "Path to the emacsclient binary";
};
debt_query = lib.mkOption {
type = lib.types.str;
default = "tag:unread and date:..1w";
description = "notmuch query whose count drives the mail-debt severity color on the bar widget";
};
debt_warn_at = lib.mkOption {
type = lib.types.int;
default = 1;
description = "Debt count at which the widget switches to the warn (orange) color";
};
debt_alarm_at = lib.mkOption {
type = lib.types.int;
default = 6;
description = "Debt count at which the widget switches to the alarm (red) color";
};
saved_searches = lib.mkOption {
type = lib.types.listOf (
lib.types.submodule {
options = {
name = lib.mkOption {
type = lib.types.str;
description = "Display label shown in the search launcher";
};
query = lib.mkOption {
type = lib.types.str;
description = "notmuch query to run when this saved search is activated";
};
};
}
);
default = [ ];
description = "Saved searches shown in the notmuch search launcher when the entry is empty";
};
};
screenrec = {
enable = lib.mkOption {
type = lib.types.bool;
default = false;
description = "Whether to enable the screenrec widget and menu";
};
output_dir = lib.mkOption {
type = lib.types.str;
default = "~/Videos/wl-screenrec";
description = "Directory to save recordings into";
};
};
power = {
lock_command = lib.mkOption {
type = lib.types.listOf lib.types.str;
default = [ "waylock" ];
description = "argv for the Lock action in the power menu";
};
};
org = {
enable = lib.mkOption {
type = lib.types.bool;
default = false;
description = "Whether to enable the org-mode agenda widget";
};
paths = lib.mkOption {
type = lib.types.listOf lib.types.str;
default = [ ];
description = "Org files / directories / globs to scan (~ is expanded)";
};
todo_keywords = lib.mkOption {
type = lib.types.listOf lib.types.str;
default = [
"TODO"
"NEXT"
"IN-PROGRESS"
];
description = "Active TODO keywords used by orgparse when no #+TODO: header is present";
};
done_keywords = lib.mkOption {
type = lib.types.listOf lib.types.str;
default = [
"DONE"
"CANCELLED"
];
description = "Done keywords (filtered out of the count)";
};
counted_states = lib.mkOption {
type = lib.types.listOf lib.types.str;
default = [
"TODO"
"NEXT"
"IN-PROGRESS"
];
description = "States summed for the bar badge count";
};
update_interval = lib.mkOption {
type = lib.types.int;
default = 60000;
description = "Polling interval in milliseconds (mtime-checked, only re-parses on change)";
};
emacsclient_command = lib.mkOption {
type = lib.types.str;
default = "emacsclient";
description = "Command used to open headlines (split on whitespace)";
};
dropdown_width = lib.mkOption {
type = lib.types.int;
default = 420;
description = "Dropdown width in pixels";
};
dropdown_height = lib.mkOption {
type = lib.types.int;
default = 480;
description = "Dropdown height in pixels";
};
};
notifications = {
enable = lib.mkOption {
type = lib.types.bool;
default = false;
description = "Whether to enable the notification toast service. Owns org.freedesktop.Notifications, so other notification daemons (mako, dunst, swaync) must be disabled.";
};
anchor = lib.mkOption {
type = lib.types.str;
default = "top center";
description = "Layer-shell anchor for the toast stack";
};
margin = lib.mkOption {
type = lib.types.str;
default = "8px";
description = "Layer-shell margin for the toast stack";
};
width = lib.mkOption {
type = lib.types.int;
default = 360;
description = "Width of each notification toast in pixels";
};
timeout_ms = lib.mkOption {
type = lib.types.int;
default = 10000;
description = "Auto-close timeout for notifications in milliseconds";
};
history_size = lib.mkOption {
type = lib.types.int;
default = 50;
description = "How many past notifications the in-memory center keeps";
};
image_max_px = lib.mkOption {
type = lib.types.int;
default = 128;
description = "Max edge in pixels for stored notification thumbnails";
};
center_width = lib.mkOption {
type = lib.types.int;
default = 380;
description = "Width of the notification center side rail in pixels";
};
settings = lib.mkOption {
type = lib.types.submodule {
options = {
vinyl = {
enable = lib.mkOption {
type = lib.types.bool;
default = false;
};
};
};
default = {
vinyl.enable = false;
battery.enable = false;
buddy.enable = false;
height = 40;
logLevel = "WARNING";
window_title.enable = true;
stylix.enable = false;
battery = {
enable = lib.mkOption {
type = lib.types.bool;
default = false;
};
};
height = lib.mkOption {
type = lib.types.int;
default = 40;
description = "Height of the status bar in pixels";
};
logLevel = lib.mkOption {
type = lib.types.enum [ "TRACE" "DEBUG" "INFO" "SUCCESS" "WARNING" "ERROR" "CRITICAL" ];
default = "WARNING";
description = "Log level for the status bar (loguru levels: TRACE, DEBUG, INFO, SUCCESS, WARNING, ERROR, CRITICAL)";
};
window_title = {
enable = lib.mkOption {
type = lib.types.bool;
default = true;
description = "Whether to show the window title in the center of the bar";
};
};
stylix = lib.mkOption {
type = lib.types.attrsOf lib.types.anything;
default = { enable = false; };
description = "Stylix configuration passed from the stylix module";
};
calendar = {
enable = true;
khal_path = "khal";
enable = lib.mkOption {
type = lib.types.bool;
default = true;
description = "Whether to enable the calendar widget";
};
khal_path = lib.mkOption {
type = lib.types.str;
default = "khal";
description = "Path to the khal binary";
};
};
notmuch = {
enable = true;
notmuch_path = "notmuch";
emacsclient_command = "emacsclient";
debt_query = "tag:unread and date:..1w";
debt_warn_at = 1;
debt_alarm_at = 6;
saved_searches = [ ];
enable = lib.mkOption {
type = lib.types.bool;
default = true;
description = "Whether to enable the notmuch email widget";
};
notmuch_path = lib.mkOption {
type = lib.types.str;
default = "notmuch";
description = "Path to the notmuch binary";
};
emacsclient_command = lib.mkOption {
type = lib.types.str;
default = "emacsclient";
description = "Path to the emacsclient binary";
};
debt_query = lib.mkOption {
type = lib.types.str;
default = "tag:unread and date:..1w";
description = "notmuch query whose count drives the mail-debt severity color on the bar widget";
};
debt_warn_at = lib.mkOption {
type = lib.types.int;
default = 1;
description = "Debt count at which the widget switches to the warn (orange) color";
};
debt_alarm_at = lib.mkOption {
type = lib.types.int;
default = 6;
description = "Debt count at which the widget switches to the alarm (red) color";
};
saved_searches = lib.mkOption {
type = lib.types.listOf (lib.types.submodule {
options = {
name = lib.mkOption {
type = lib.types.str;
description = "Display label shown in the search launcher";
};
query = lib.mkOption {
type = lib.types.str;
description = "notmuch query to run when this saved search is activated";
};
};
});
default = [ ];
description = "Saved searches shown in the notmuch search launcher when the entry is empty";
};
};
screenrec = {
enable = false;
output_dir = "~/Videos/wl-screenrec";
enable = lib.mkOption {
type = lib.types.bool;
default = false;
description = "Whether to enable the screenrec widget and menu";
};
output_dir = lib.mkOption {
type = lib.types.str;
default = "~/Videos/wl-screenrec";
description = "Directory to save recordings into";
};
};
power = {
lock_command = [ "waylock" ];
};
org = {
enable = false;
paths = [ ];
todo_keywords = [
"TODO"
"NEXT"
"IN-PROGRESS"
];
done_keywords = [
"DONE"
"CANCELLED"
];
counted_states = [
"TODO"
"NEXT"
"IN-PROGRESS"
];
update_interval = 60000;
emacsclient_command = "emacsclient";
dropdown_width = 420;
dropdown_height = 480;
lock_command = lib.mkOption {
type = lib.types.listOf lib.types.str;
default = [ "waylock" ];
description = "argv for the Lock action in the power menu";
};
};
notifications = {
enable = false;
anchor = "top center";
margin = "8px";
width = 360;
timeout_ms = 10000;
history_size = 50;
image_max_px = 128;
center_width = 380;
enable = lib.mkOption {
type = lib.types.bool;
default = false;
description = "Whether to enable the notification toast service. Owns org.freedesktop.Notifications, so other notification daemons (mako, dunst, swaync) must be disabled.";
};
anchor = lib.mkOption {
type = lib.types.str;
default = "top center";
description = "Layer-shell anchor for the toast stack";
};
margin = lib.mkOption {
type = lib.types.str;
default = "8px";
description = "Layer-shell margin for the toast stack";
};
width = lib.mkOption {
type = lib.types.int;
default = 360;
description = "Width of each notification toast in pixels";
};
timeout_ms = lib.mkOption {
type = lib.types.int;
default = 10000;
description = "Auto-close timeout for notifications in milliseconds";
};
history_size = lib.mkOption {
type = lib.types.int;
default = 50;
description = "How many past notifications the in-memory center keeps";
};
image_max_px = lib.mkOption {
type = lib.types.int;
default = 128;
description = "Max edge in pixels for stored notification thumbnails";
};
center_width = lib.mkOption {
type = lib.types.int;
default = 380;
description = "Width of the notification center side rail in pixels";
};
};
};
};
};
config = lib.mkIf config.services.sims.enable {
systemd.user.services.sims =
let
configFile = settingsFormat.generate "config.yaml" cfg.settings;
in
{
Unit = {
Description = "sims status bar";
# Start as part of the graphical session rather than at
# user-manager boot. The compositor populates the systemd
# user environment (WAYLAND_DISPLAY, NIXOS_OZONE_WL, ...)
# via `import-environment` before it activates
# graphical-session.target, so binding here guarantees sims
# — and every app it launches via `systemd-run --scope` —
# inherits the correct Wayland environment.
PartOf = [ "graphical-session.target" ];
After = [ "graphical-session.target" ];
};
Service = {
ExecStart = "${config.services.sims.package}/bin/sims --config ${configFile}";
Restart = "on-failure";
};
Install = {
WantedBy = [ "graphical-session.target" ];
};
default = {
vinyl.enable = false;
battery.enable = false;
height = 40;
logLevel = "WARNING";
window_title.enable = true;
stylix.enable = false;
calendar = {
enable = true;
khal_path = "khal";
};
notmuch = {
enable = true;
notmuch_path = "notmuch";
emacsclient_command = "emacsclient";
debt_query = "tag:unread and date:..1w";
debt_warn_at = 1;
debt_alarm_at = 6;
saved_searches = [ ];
};
screenrec = {
enable = false;
output_dir = "~/Videos/wl-screenrec";
};
power = {
lock_command = [ "waylock" ];
};
notifications = {
enable = false;
anchor = "top center";
margin = "8px";
width = 360;
timeout_ms = 10000;
history_size = 50;
image_max_px = 128;
center_width = 380;
};
};
};
};
config = lib.mkIf config.services.sims.enable {
systemd.user.services.sims =
let
configFile = settingsFormat.generate "config.yaml" cfg.settings;
in
{
Unit = {
Description = "sims status bar";
After = [ "graphical-session.target" ];
};
Service = {
ExecStart = "${config.services.sims.package}/bin/sims --config ${configFile}";
Restart = "on-failure";
};
Install = {
WantedBy = [ "default.target" ];
};
};
};
};
stylix-sims = import ./nix/stylix/hm.nix;
};
};
-1
View File
@@ -50,7 +50,6 @@ python3Packages.buildPythonApplication {
pywayland
pyyaml
platformdirs
orgparse
];
doCheck = false;
dontWrapGApps = true;
-1
View File
@@ -44,7 +44,6 @@ pkgs.mkShell {
python-lsp-ruff
pyyaml
platformdirs
orgparse
]
))
];
Executable → Regular
View File
-10
View File
@@ -76,13 +76,3 @@ NOTIFICATIONS = app_config.get("notifications", {
BAR_HEIGHT = app_config.get("height", 40)
LOG_LEVEL = app_config.get("logLevel", "WARNING")
DEV = app_config.get("dev", False)
BUDDY = app_config.get("buddy", {"enable": False})
ORG = app_config.get("org", {"enable": False})
ORG.setdefault("paths", [])
ORG.setdefault("todo_keywords", ["TODO", "NEXT", "IN-PROGRESS"])
ORG.setdefault("done_keywords", ["DONE", "CANCELLED"])
ORG.setdefault("counted_states", ["TODO", "NEXT", "IN-PROGRESS"])
ORG.setdefault("update_interval", 60_000)
ORG.setdefault("emacsclient_command", "emacsclient")
ORG.setdefault("dropdown_width", 420)
ORG.setdefault("dropdown_height", 480)
+1 -15
View File
@@ -29,11 +29,9 @@ from .modules.launcher.screenshot import ScreenshotMenu
from .modules.calendar import CalendarService
from .modules.control_center import ControlCenter
from .modules.notifications import NotificationToasts
from .modules.org import OrgDropdown
from .modules.stylix import get_stylix_css_path
from .modules.vinyl import VinylButton
from .services.org import OrgService
from .config import CALENDAR, NOTIFICATIONS, ORG, POWER, SCREENREC, STYLIX, VINYL
from .config import CALENDAR, NOTIFICATIONS, POWER, SCREENREC, STYLIX, VINYL
from .services.fenster import get_i3_connection
from .services.notification_history import NotificationHistoryService
from .services.screenrec import ScreenrecService
@@ -94,14 +92,6 @@ if notification_history is not None:
width=NOTIFICATIONS.get("center_width", 380),
)
org_service: OrgService | None = None
org_dropdown: OrgDropdown | None = None
if ORG.get("enable", False):
org_service = OrgService(
update_interval=int(ORG.get("update_interval", 60_000)),
)
org_dropdown = OrgDropdown(service=org_service, monitor=0)
bar_windows = []
notmuch_widget = None
@@ -112,8 +102,6 @@ if notification_toasts is not None:
_app_windows.append(notification_toasts)
if control_center is not None:
_app_windows.append(control_center)
if org_dropdown is not None:
_app_windows.append(org_dropdown)
app = Application("sims", *_app_windows)
@@ -241,8 +229,6 @@ def spawn_bars():
monitor=i,
screenrec_service=screenrec_service if i == 0 else None,
control_center=control_center if i == 0 else None,
org_service=org_service,
org_dropdown=org_dropdown if i == 0 else None,
)
bar_windows.append(bar)
if i == 0 and bar.notmuch:
+9 -33
View File
@@ -8,21 +8,17 @@ from sims.modules.player import Player, PlayerSmall
from sims.modules.battery import Battery
from sims.modules.control_center import ControlCenter
from sims.modules.notmuch import NotmuchWidget
from sims.modules.buddy import Buddy
from sims.modules.org import OrgDropdown, OrgWidget
from sims.modules.screenrec import ScreenrecWidget
from sims.modules.microphone import MicrophoneWidget
from fabric.widgets.wayland import WaylandWindow as Window
from fabric.system_tray.widgets import SystemTray
from sims.widgets.fenster import FensterWorkspaces, FensterActiveWindow
from sims.services.org import OrgService
from sims.services.screenrec import ScreenrecService
from sims.services.smart_corners import get_smart_corners_service
from fabric.widgets.button import Button
from fabric.widgets.circularprogressbar import CircularProgressBar
from sims.services.system_stats import get_system_stats_service
from sims.services.system_stats import SystemStatsService
from sims.config import BATTERY, BAR_HEIGHT, WINDOW_TITLE, NOTMUCH, BUDDY, ORG
from sims.config import BATTERY, BAR_HEIGHT, WINDOW_TITLE, NOTMUCH
class StatusBar(Window):
@@ -33,8 +29,6 @@ class StatusBar(Window):
monitor: int = 1,
screenrec_service: ScreenrecService | None = None,
control_center: ControlCenter | None = None,
org_service: OrgService | None = None,
org_dropdown: OrgDropdown | None = None,
):
super().__init__(
name="sims",
@@ -98,33 +92,19 @@ class StatusBar(Window):
if NOTMUCH["enable"]:
self.notmuch = NotmuchWidget()
self.buddy = None
if BUDDY.get("enable", False):
notmuch_service = self.notmuch.service if self.notmuch is not None else None
self.buddy = Buddy(notmuch_service=notmuch_service)
self.screenrec = None
if screenrec_service is not None:
self.screenrec = ScreenrecWidget(screenrec_service)
self.org = None
if org_service is not None:
self.org = OrgWidget(service=org_service, dropdown=org_dropdown)
self.microphone = MicrophoneWidget()
self.status_container = Box(
name="widgets-container",
spacing=4,
orientation="h",
children=[self.microphone, self.progress_bars_overlay],
children=self.progress_bars_overlay,
)
end_container_children = []
if self.org is not None:
end_container_children.append(self.org)
end_container_children.append(self.status_container)
if self.system_tray:
end_container_children.append(self.system_tray)
@@ -144,21 +124,17 @@ class StatusBar(Window):
if WINDOW_TITLE["enable"]:
center_children.append(self.active_window)
start_container_children = [
Image(name="nixos-label", icon_name="nix-snowflake-white", icon_size=20),
self.workspaces,
self.player_small,
]
if self.buddy is not None:
start_container_children.append(self.buddy)
self.inner = CenterBox(
name="sims-inner",
start_children=Box(
name="start-container",
spacing=6,
orientation="h",
children=start_container_children,
children=[
Image(name="nixos-label", icon_name="nix-snowflake-white", icon_size=20),
self.workspaces,
self.player_small,
],
),
center_children=Box(
name="center-container",
@@ -175,7 +151,7 @@ class StatusBar(Window):
)
self.children = self.inner
self.system_stats_service = get_system_stats_service()
self.system_stats_service = SystemStatsService(update_interval=3000)
self.system_stats_service.connect("stats-changed", self.update_progress_bars)
self.set_size_request(-1, BAR_HEIGHT)
+3 -4
View File
@@ -2,7 +2,7 @@ from gi.repository import GLib
from fabric.widgets.box import Box
from fabric.widgets.label import Label
from fabric.widgets.image import Image
from sims.services.battery import get_battery_service
from sims.services.battery import BatteryService
class Battery(Box):
@@ -15,9 +15,8 @@ class Battery(Box):
self.bat_label = Label(name="bat-label", label="100%")
# Shared battery service with signal-based updates (one poller for the
# whole process, including the buddy's battery checks)
self.battery_service = get_battery_service()
# Create battery service with signal-based updates
self.battery_service = BatteryService(update_interval=10000) # Check every 10 seconds
self.battery_service.connect("battery-changed", self.update_battery)
self.children = [self.bat_icon, self.bat_label]
-23
View File
@@ -1,23 +0,0 @@
from fabric.widgets.box import Box
from sims.widgets.buddy import BuddyWidget
from sims.services.buddy import get_buddy_service
class Buddy(Box):
def __init__(self, notmuch_service=None, **kwargs):
super().__init__(name="buddy-container", v_align="center", **kwargs)
self.service = get_buddy_service(notmuch_service=notmuch_service)
self.sprite = BuddyWidget(on_pet=self._on_pet)
self.service.connect("mood-changed", self._on_mood_changed)
self.children = [self.sprite]
# The shared service only emits on *changes* and may already sit in a
# non-idle mood when this output's sprite is built: adopt it now.
self.sprite.update_mood(self.service.mood)
def _on_mood_changed(self, _service, mood: str):
self.sprite.update_mood(mood)
def _on_pet(self, **kwargs):
if kwargs.get("petted"):
self.service.pet()
+17 -27
View File
@@ -1,3 +1,4 @@
import json
import os
import subprocess
import shutil
@@ -130,19 +131,18 @@ class CalendarService:
return
try:
# khal 0.11.3 does not support the newer `--json` output flags.
# Use the documented `list --format ... --day-format ... START DELTA`
# form instead and parse one event per line. `{bell}` is an uncommon
# control-code delimiter supported by khal's formatter.
cmd = [
khal_path,
"list",
"--format",
"{start-date}{bell}{start-time}{bell}{end-date}{bell}{end-time}{bell}{title}{bell}{location}",
"--day-format",
"",
"--json",
"title",
"--json",
"start",
"--json",
"end",
"--json",
"location",
"today",
"eod",
]
logger.info(f"[Calendar] Running command: {' '.join(cmd)}")
result = subprocess.run(
@@ -155,26 +155,16 @@ class CalendarService:
logger.info(f"[Calendar] Command stderr: {result.stderr[:200]}...")
if result.stdout.strip():
lines = result.stdout.strip().split("\n")
all_events = []
for line in result.stdout.splitlines():
if not line.strip():
continue
parts = line.split("\a", 5)
if len(parts) != 6:
logger.warning(f"[Calendar] Could not parse khal output line: {line}")
continue
start_date, start_time, end_date, end_time, title, location = parts
all_events.append(
{
"title": title.strip() or "No title",
"start": f"{start_date} {start_time}".strip() if start_time else "",
"end": f"{end_date} {end_time}".strip() if end_time else "",
"location": location.strip(),
}
)
for line in lines:
if line.strip():
try:
events = json.loads(line)
all_events.extend(events)
except json.JSONDecodeError:
continue
self.events = all_events
logger.info(f"[Calendar] Found {len(self.events)} events using subprocess")
+9 -51
View File
@@ -1,7 +1,6 @@
import re
import shlex
import subprocess
import time
from fabric.utils.helpers import DesktopApp, get_desktop_applications
from fabric.widgets.box import Box
@@ -10,8 +9,6 @@ from fabric.widgets.label import Label
from gi.repository import Gtk
from .base import FuzzyMenu
from .history import LaunchHistory
from .ranking import rank_items
ICON_SIZE = 32
@@ -20,27 +17,14 @@ _FIELD_CODE_RE = re.compile(r"^%[fFuUickdDnNvm]$")
class AppProvider:
def __init__(self, history: LaunchHistory | None = None):
self._history = history if history is not None else LaunchHistory()
def items(self) -> list[DesktopApp]:
return get_desktop_applications()
def filter(self, items: list[DesktopApp], query: str) -> list[DesktopApp]:
"""Rank apps against *query*, best match first.
With an empty query everything matches equally, so the order is
purely "what gets launched most" - the launcher opens on your
most useful apps instead of an arbitrary alphabetical slice.
"""
now = time.time()
return rank_items(
items,
query,
_fields,
bonus_of=lambda app: self._history.bonus(_app_key(app), now=now),
tie_break_of=_sort_name,
)
if not query:
return items
q = query.lower()
return [a for a in items if _matches(a, q)]
def render(self, item: DesktopApp) -> Gtk.Widget:
children: list[Gtk.Widget] = []
@@ -61,7 +45,6 @@ class AppProvider:
return Box(name="slot-box", orientation="h", spacing=10, children=children)
def activate(self, item: DesktopApp) -> None:
self._history.record(_app_key(item))
# Launch in a transient systemd --user scope so the app gets its own
# cgroup instead of inheriting sims.service's. start_new_session alone
# only changes POSIX session/pgid; systemd tracks units by cgroup and
@@ -87,36 +70,11 @@ class AppProvider:
item.launch()
def _fields(app: DesktopApp) -> dict[str, str | None]:
return {
"name": app.name,
"display_name": app.display_name,
"generic_name": app.generic_name,
"executable": app.executable,
}
def _sort_name(app: DesktopApp) -> str:
return (app.display_name or app.name or "").casefold()
def _app_key(app: DesktopApp) -> str:
"""Stable identity for the launch history.
DesktopApp does not expose the desktop file id, but it wraps the
GioUnix.DesktopAppInfo which does. Fall back to the command line
(unique per Steam game) and finally the name.
"""
info = getattr(app, "_app", None)
get_id = getattr(info, "get_id", None) if info is not None else None
if callable(get_id):
try:
app_id = get_id()
except Exception: # pragma: no cover - defensive
app_id = None
if isinstance(app_id, str) and app_id:
return app_id
return app.command_line or app.executable or app.name or ""
def _matches(app: DesktopApp, q: str) -> bool:
for field in (app.name, app.display_name, app.generic_name, app.executable):
if field and q in field.lower():
return True
return False
def AppLauncher(monitor: int = 0) -> FuzzyMenu:
+4 -11
View File
@@ -9,8 +9,6 @@ from gi.repository import Gdk, Gtk
from sims.services.fenster import focused_output_index
from .ranking import rank_items
class LauncherProvider(Protocol):
def items(self) -> list[Any]: ...
@@ -49,15 +47,10 @@ class StaticActionProvider:
return list(self._static or [])
def filter(self, items: list[StaticAction], query: str) -> list[StaticAction]:
# Fixed menus (screenshot, power, screenrec) have no history: an
# empty query keeps the declaration order, a real query is ranked
# like everything else in the launcher.
return rank_items(
items,
query,
lambda action: {"label": action.label},
weights={"label": 1.0},
)
if not query:
return items
q = query.lower()
return [i for i in items if q in i.label.lower()]
def render(self, item: StaticAction) -> Gtk.Widget:
return Box(
+4 -9
View File
@@ -6,7 +6,6 @@ from fabric.widgets.label import Label
from gi.repository import Gtk
from .base import FuzzyMenu
from .ranking import rank_items
@dataclass
@@ -35,14 +34,10 @@ class ClipboardProvider:
return entries
def filter(self, items: list[ClipEntry], query: str) -> list[ClipEntry]:
# Empty query keeps cliphist's newest-first order; a query is
# ranked like every other launcher menu.
return rank_items(
items,
query,
lambda entry: {"preview": entry.preview},
weights={"preview": 1.0},
)
if not query:
return items
q = query.lower()
return [e for e in items if q in e.preview.lower()]
def render(self, item: ClipEntry) -> Gtk.Widget:
return Box(
-167
View File
@@ -1,167 +0,0 @@
"""Persistent launch history for the app launcher.
Remembers how often and how recently each application was launched so
``AppProvider`` can rank frequently used apps first (frecency, the
same frequency + recency idea as zoxide). The file lives in the system
cache directory - losing it only resets the ranking, never breaks the
launcher.
Layout (JSON, written atomically)::
{"version": 1, "entries": {"steam.desktop": {"count": 12, "last_used": 1712345678.0}}}
"""
from __future__ import annotations
import json
import os
import tempfile
import time
from dataclasses import dataclass
from pathlib import Path
from loguru import logger
from platformdirs import user_cache_dir
from .ranking import frecency, popularity_bonus
CACHE_VERSION = 1
MAX_ENTRIES = 500
MAX_AGE_DAYS = 365.0
DEFAULT_HALF_LIFE_DAYS = 30.0
def default_cache_path() -> Path:
return Path(user_cache_dir("sims")) / "launcher-history.json"
@dataclass
class _Entry:
count: int = 0
last_used: float = 0.0
class LaunchHistory:
"""Launch counts/recency, keyed by launcher-specific item id."""
def __init__(
self,
path: Path | str | None = None,
*,
half_life_days: float = DEFAULT_HALF_LIFE_DAYS,
):
self.path = Path(path) if path is not None else default_cache_path()
self._half_life_days = half_life_days
self._entries: dict[str, _Entry] = self._load()
def record(self, key: str, *, now: float | None = None) -> None:
"""Count one launch of *key* and persist the history."""
if not key:
return
now = time.time() if now is None else now
entry = self._entries.get(key)
if entry is None:
entry = self._entries[key] = _Entry()
entry.count += 1
entry.last_used = now
self._save()
def bonus(
self,
key: str,
*,
now: float | None = None,
max_bonus: float = 150.0,
) -> float:
"""Bounded popularity bonus for *key*, 0.0 when unknown."""
entry = self._entries.get(key)
if entry is None or entry.count <= 0:
return 0.0
now = time.time() if now is None else now
score = frecency(
entry.count,
entry.last_used,
now,
half_life_days=self._half_life_days,
)
return popularity_bonus(score, max_bonus=max_bonus)
# -- persistence ------------------------------------------------------
def _load(self) -> dict[str, _Entry]:
try:
with open(self.path, "r", encoding="utf-8") as handle:
data = json.load(handle)
except FileNotFoundError:
return {}
except (OSError, ValueError) as exc:
logger.warning("launcher history {} unreadable: {}", self.path, exc)
return {}
raw_entries = data.get("entries") if isinstance(data, dict) else None
if not isinstance(raw_entries, dict):
logger.warning("launcher history {}: unexpected format", self.path)
return {}
entries: dict[str, _Entry] = {}
for key, raw in raw_entries.items():
if not isinstance(raw, dict):
continue
try:
count = int(raw.get("count", 0))
last_used = float(raw.get("last_used", 0.0))
except (TypeError, ValueError):
continue
if count > 0:
entries[key] = _Entry(count=count, last_used=last_used)
return self._prune(entries, time.time())
def _prune(self, entries: dict[str, _Entry], now: float) -> dict[str, _Entry]:
cutoff = now - MAX_AGE_DAYS * 86_400.0
kept = {
key: entry
for key, entry in entries.items()
if entry.count > 0 and entry.last_used >= cutoff
}
if len(kept) > MAX_ENTRIES:
top = sorted(
kept.items(),
key=lambda item: frecency(
item[1].count,
item[1].last_used,
now,
half_life_days=self._half_life_days,
),
reverse=True,
)[:MAX_ENTRIES]
kept = dict(top)
return kept
def _save(self) -> None:
self._entries = self._prune(self._entries, time.time())
payload = {
"version": CACHE_VERSION,
"entries": {
key: {"count": entry.count, "last_used": entry.last_used}
for key, entry in sorted(self._entries.items())
},
}
temp_path: str | None = None
try:
self.path.parent.mkdir(parents=True, exist_ok=True)
fd, temp_path = tempfile.mkstemp(
dir=self.path.parent, prefix=self.path.name, suffix=".tmp"
)
with os.fdopen(fd, "w", encoding="utf-8") as handle:
json.dump(payload, handle)
os.replace(temp_path, self.path)
temp_path = None
except OSError as exc:
logger.warning("could not write launcher history {}: {}", self.path, exc)
finally:
if temp_path is not None:
try:
os.unlink(temp_path)
except OSError:
pass
+2 -2
View File
@@ -108,7 +108,7 @@ class NotmuchSearchProvider:
return hits
def render_hit(self, hit: NotmuchHit) -> Gtk.Widget:
text = Box(name="notmuch-text", orientation="v", spacing=0, h_expand=True)
text = Box(name="notmuch-text", orientation="v", spacing=0)
text.add(
Label(
label=hit.subject,
@@ -138,7 +138,7 @@ class NotmuchSearchProvider:
)
def render_saved(self, item: SavedSearch) -> Gtk.Widget:
text = Box(name="notmuch-text", orientation="v", spacing=0, h_expand=True)
text = Box(name="notmuch-text", orientation="v", spacing=0)
text.add(
Label(label=item.name, name="notmuch-saved-name", h_align="start")
)
-263
View File
@@ -1,263 +0,0 @@
"""Search ranking for launcher items.
The launcher used to keep items in provider order (alphabetical for
desktop apps) and filter by plain substring. That breaks as soon as a
query matches several apps through different fields: typing ``steam``
listed every installed Steam game (each of their desktop files has
``Exec=steam steam://rungameid/...``) in the middle of the list, with
Steam itself somewhere in between.
This module scores a query against a set of named fields:
* every whitespace-separated token must match somewhere (AND semantics);
* match *kind* dominates the score:
exact > prefix > word start > substring > subsequence;
* fields are weighted so a match on the user-visible name outranks the
same kind of match on the generic name or the executable;
* a bounded bonus can be added on top (launch frecency, see
``history.py``). The bonus is small enough that it can never beat a
better match kind on the same field.
The score is only used for ordering, so the absolute numbers do not
matter - only the gaps between them.
"""
from __future__ import annotations
from dataclasses import dataclass
from enum import IntEnum
from typing import Any, Callable, Mapping, Sequence, TypeVar
# Fuzzy (subsequence) matching on tiny queries matches almost anything,
# so only use it once the query is long enough to be meaningful.
MIN_SUBSEQUENCE_LEN = 3
Item = TypeVar("Item")
class MatchKind(IntEnum):
NONE = 0
SUBSEQUENCE = 1
SUBSTRING = 2
WORD_PREFIX = 3
PREFIX = 4
EXACT = 5
# Base score per match kind. The gaps are much larger than any quality
# or popularity bonus, so a better kind of match on the same field always
# wins over a bonus-carrying worse one.
_KIND_SCORE: Mapping[MatchKind, float] = {
MatchKind.EXACT: 1000.0,
MatchKind.PREFIX: 650.0,
MatchKind.WORD_PREFIX: 450.0,
MatchKind.SUBSTRING: 250.0,
MatchKind.SUBSEQUENCE: 90.0,
}
# Weight of the match *quality* (max 75 points total):
_POSITION_WEIGHT = 25.0 # earlier in the field is better
_COVERAGE_WEIGHT = 25.0 # token covers more of the field is better
_CONTIGUITY_WEIGHT = 25.0 # subsequence: fewer skipped chars is better
# How much each desktop-entry field counts. The name must always beat
# generic-name/executable matches, otherwise every Steam game (Exec=
# "steam steam://rungameid/<id>") drowns out Steam itself.
DEFAULT_FIELD_WEIGHTS: Mapping[str, float] = {
"name": 1.0,
"display_name": 0.95,
"generic_name": 0.45,
"executable": 0.55,
}
@dataclass(frozen=True)
class FieldMatch:
"""Where and how a token matched a single field."""
kind: MatchKind
position: int
span: int
def quality(self, value_len: int, token_len: int) -> float:
value_len = max(value_len, 1)
position = 1.0 - self.position / value_len
coverage = token_len / value_len
contiguity = token_len / max(self.span, 1)
return (
_POSITION_WEIGHT * position
+ _COVERAGE_WEIGHT * coverage
+ _CONTIGUITY_WEIGHT * contiguity
)
def _is_word_start(value: str, position: int) -> bool:
return position == 0 or not value[position - 1].isalnum()
def _subsequence_span(value: str, token: str) -> tuple[int, int] | None:
"""Greedy subsequence match; returns (start, span) or None."""
first = -1
last = -1
index = 0
for position, char in enumerate(value):
if char == token[index]:
if first == -1:
first = position
last = position
index += 1
if index == len(token):
return first, last - first + 1
return None
def find_match(value: str, token: str) -> FieldMatch | None:
"""Best match of *token* inside *value* (both lowercased here).
Returns None when the token does not occur in the field at all.
"""
if not value or not token:
return None
value = value.lower()
token = token.lower()
if value == token:
return FieldMatch(MatchKind.EXACT, 0, len(token))
if value.startswith(token):
return FieldMatch(MatchKind.PREFIX, 0, len(token))
position = value.find(token)
if position != -1:
kind = (
MatchKind.WORD_PREFIX
if _is_word_start(value, position)
else MatchKind.SUBSTRING
)
return FieldMatch(kind, position, len(token))
if len(token) >= MIN_SUBSEQUENCE_LEN:
span = _subsequence_span(value, token)
if span is not None:
return FieldMatch(MatchKind.SUBSEQUENCE, span[0], span[1])
return None
def score_token(
token: str,
fields: Mapping[str, str | None],
weights: Mapping[str, float] = DEFAULT_FIELD_WEIGHTS,
) -> float | None:
"""Best weighted score of *token* across *fields*, or None."""
best: float | None = None
for field, weight in weights.items():
value = fields.get(field)
if not value:
continue
match = find_match(value, token)
if match is None:
continue
score = weight * (
_KIND_SCORE[match.kind] + match.quality(len(value), len(token))
)
if best is None or score > best:
best = score
return best
def score_query(
query: str,
fields: Mapping[str, str | None],
weights: Mapping[str, float] = DEFAULT_FIELD_WEIGHTS,
) -> float | None:
"""Score an item for *query*: None unless every token matches.
An empty query matches everything with score 0.0, which lets callers
fall back to time/frequency based ordering.
"""
tokens = query.split()
if not tokens:
return 0.0
total = 0.0
for token in tokens:
token_score = score_token(token, fields, weights)
if token_score is None:
return None
total += token_score
return total / len(tokens)
def rank_items(
items: Sequence[Item],
query: str,
fields_of: Callable[[Item], Mapping[str, str | None]],
*,
weights: Mapping[str, float] = DEFAULT_FIELD_WEIGHTS,
bonus_of: Callable[[Item], float] | None = None,
tie_break_of: Callable[[Item], Any] | None = None,
) -> list[Item]:
"""Rank *items* by how well they match *query*, best first.
This is the shared filter for every launcher provider: apps, windows,
clipboard entries and static action menus all describe their searchable
text as weighted fields via *fields_of*.
With an empty query the provider order is kept, unless *bonus_of* is
given - then items are ordered by bonus instead (used by the app
launcher to show launch frecency). *bonus_of* is added to the match
score and must stay small enough not to outrank a better match kind.
Ties keep provider order, or fall back to *tie_break_of* when given.
"""
if not query.split():
if bonus_of is None:
return list(items)
scored = [(bonus_of(item), item) for item in items]
else:
scored = []
for item in items:
score = score_query(query, fields_of(item), weights)
if score is None:
continue
if bonus_of is not None:
score += bonus_of(item)
scored.append((score, item))
if tie_break_of is None:
scored.sort(key=lambda pair: -pair[0])
else:
scored.sort(key=lambda pair: (-pair[0], tie_break_of(pair[1])))
return [item for _, item in scored]
def frecency(
count: int,
last_used: float | None,
now: float,
*,
half_life_days: float = 30.0,
) -> float:
"""Launch count decayed by age with the given half-life."""
if count <= 0:
return 0.0
if last_used is None:
return float(count)
age_days = max(0.0, (now - last_used) / 86_400.0)
return count * 0.5 ** (age_days / max(half_life_days, 1e-6))
def popularity_bonus(
frecency_score: float,
*,
max_bonus: float = 150.0,
half_bonus_at: float = 5.0,
) -> float:
"""Saturating, monotonic map of frecency into ``[0, max_bonus)``.
Keeps the launch-history signal bounded so it can reorder equal
matches but never beat a better match kind.
"""
if frecency_score <= 0.0:
return 0.0
return max_bonus * frecency_score / (frecency_score + half_bonus_at)
+8 -20
View File
@@ -3,15 +3,6 @@ from fabric.widgets.box import Box
from fabric.widgets.label import Label
from gi.repository import Gtk
from .ranking import rank_items
# A window title is what people search for; the app id is a fallback.
# Weighted below 0.65 on purpose: an exact app-id match must not beat a
# window whose *title* starts with the query, so a query like "firefox"
# still orders the Firefox windows by their titles.
_WINDOW_WEIGHTS = {"title": 1.0, "app_id": 0.6}
class WindowProvider:
def items(self) -> list[dict]:
@@ -43,17 +34,14 @@ class WindowProvider:
return windows
def filter(self, items: list[dict], query: str) -> list[dict]:
# No history for windows - empty query keeps workspace order,
# typed queries are ranked by title/app id match quality.
return rank_items(
items,
query,
lambda window: {
"title": window.get("title", ""),
"app_id": window.get("app_id", ""),
},
weights=_WINDOW_WEIGHTS,
)
if not query:
return items
q = query.lower()
return [
w for w in items
if q in w.get("title", "").lower()
or q in w.get("app_id", "").lower()
]
def render(self, item: dict) -> Gtk.Widget:
title = item.get("title", "")
-68
View File
@@ -1,68 +0,0 @@
from fabric.widgets.box import Box
from fabric.widgets.button import Button
from fabric.widgets.label import Label
from sims.services.audio import get_audio_service
MIC_ICON = "󰍬"
MIC_MUTED_ICON = "󰍭"
class MicrophoneWidget(Box):
"""Bar widget showing microphone state; click to toggle mute.
Mutes *all* input sources (not just the default one) so virtual
sources like EasyEffects and their underlying hardware mics are
silenced together.
"""
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.audio = get_audio_service()
self.icon_label = Label(
name="microphone-icon",
label=MIC_ICON,
)
self.button = Button(
name="microphone-button",
child=self.icon_label,
on_clicked=self.on_clicked,
style="background: transparent; border: none; margin: 0; box-shadow: none;",
)
self.add(self.button)
# `changed` fires for any stream change (incl. mute state) and
# `microphone-changed` when the default source switches.
self.audio.connect("changed", self.update_icon)
self.audio.connect("microphone-changed", self.update_icon)
self.update_icon()
def on_clicked(self, *_):
mics = self.audio.microphones or []
if not mics and self.audio.microphone:
mics = [self.audio.microphone]
if not mics:
return
# if anything is live -> mute everything; else unmute everything
mute = self._any_unmuted(mics)
for mic in mics:
mic.muted = mute
self.update_icon()
def _any_unmuted(self, mics) -> bool:
return any(not mic.muted for mic in mics)
def update_icon(self, *_):
mics = self.audio.microphones or []
if not mics and self.audio.microphone:
mics = [self.audio.microphone]
if not mics:
self.set_visible(False)
return
self.set_visible(True)
if self._any_unmuted(mics):
self.icon_label.set_label(MIC_ICON)
self.icon_label.remove_style_class("muted")
else:
self.icon_label.set_label(MIC_MUTED_ICON)
self.icon_label.add_style_class("muted")
+1 -17
View File
@@ -155,7 +155,7 @@ class NotmuchWidget(Button):
)
# Initialize the service
self.service = get_notmuch_service()
self.service = NotmuchService()
self.service.connect("counts-changed", self.update_display)
logger.info("[Notmuch] Notmuch widget initialized")
@@ -206,19 +206,3 @@ class NotmuchWidget(Button):
logger.info(
f"[Notmuch] Updated display: {unread} unread, {debt} aging — classes={classes}"
)
_service: "NotmuchService | None" = None
def get_notmuch_service(update_interval: int = 60000) -> "NotmuchService":
"""Return the process-wide notmuch service, creating it on first use.
Each poll shells out to `notmuch count` twice. Every bar widget used to own
a service, so one `refresh-notmuch` (or one timer tick) opened the database
2 x bars times over; now all widgets share a single poller.
"""
global _service
if _service is None:
_service = NotmuchService(update_interval=update_interval)
return _service
-283
View File
@@ -1,283 +0,0 @@
"""Org-agenda bar widget + dropdown popup.
Bar slot is a small button showing an icon + count of open items. Clicking
toggles a layer-shell dropdown listing items grouped by state. Clicking an
item opens emacsclient at the headline's line.
"""
from __future__ import annotations
import os
import subprocess
from fabric.widgets.box import Box
from fabric.widgets.button import Button
from fabric.widgets.image import Image
from fabric.widgets.label import Label
from fabric.widgets.scrolledwindow import ScrolledWindow
from fabric.widgets.wayland import WaylandWindow as Window
from gi.repository import Gdk
from loguru import logger
from sims.config import BAR_HEIGHT, ORG
from sims.services.org import OrgItem, OrgService, OrgSnapshot
# Order in which states are shown / counted. Anything else discovered at parse
# time is appended after these.
DEFAULT_STATE_ORDER = ["NEXT", "IN-PROGRESS", "TODO"]
def _state_order(states: list[str]) -> list[str]:
seen = set()
ordered: list[str] = []
for s in DEFAULT_STATE_ORDER:
if s in states and s not in seen:
ordered.append(s)
seen.add(s)
for s in states:
if s not in seen:
ordered.append(s)
seen.add(s)
return ordered
def _open_in_emacs(item: OrgItem) -> None:
cmd_template = ORG.get("emacsclient_command", "emacsclient")
parts = cmd_template.split() if isinstance(cmd_template, str) else list(cmd_template)
cmd = [*parts, "-c", f"+{item.line}", item.file]
try:
subprocess.Popen(cmd, start_new_session=True)
logger.info(f"[Org] Opened {item.file}:{item.line} in emacsclient")
except Exception as e:
logger.error(f"[Org] Failed to open emacsclient: {e}")
class OrgDropdown(Window):
def __init__(self, service: OrgService, monitor: int = 0):
width = int(ORG.get("dropdown_width", 420))
super().__init__(
name="org-dropdown",
layer="top",
anchor="top",
margin=f"{BAR_HEIGHT}px 0 0 0",
keyboard_mode="on-demand",
exclusivity="none",
visible=False,
monitor=monitor,
)
self._service = service
self._width = width
self._header = Label(
name="org-dropdown-title",
label="Org Agenda",
h_align="start",
)
close_button = Button(
name="org-dropdown-close",
image=Image(icon_name="window-close-symbolic", icon_size=16),
on_clicked=lambda *_: self.hide(),
)
header_row = Box(name="org-dropdown-header", orientation="h", spacing=8)
header_row.pack_start(self._header, True, True, 0)
header_row.pack_end(close_button, False, False, 0)
self._sections_box = Box(
name="org-dropdown-sections",
orientation="v",
spacing=10,
h_expand=True,
)
scroll = ScrolledWindow(
name="org-dropdown-scroll",
h_scrollbar_policy="never",
v_scrollbar_policy="automatic",
child=self._sections_box,
h_expand=True,
v_expand=True,
)
body = Box(
name="org-dropdown-body",
orientation="v",
spacing=8,
children=[header_row, scroll],
h_expand=True,
v_expand=True,
)
body.set_size_request(self._width, int(ORG.get("dropdown_height", 480)))
self.add(body)
self.connect("key-press-event", self._on_key_press)
self._service.connect("items-changed", lambda _s, snap: self._refresh(snap))
self._refresh(self._service.snapshot)
def toggle(self) -> None:
if self.get_visible():
self.hide()
else:
self.show()
def show(self) -> None: # type: ignore[override]
super().show()
self.show_all()
def _on_key_press(self, _w, event):
if event.keyval == Gdk.KEY_Escape:
self.hide()
return True
return False
def _refresh(self, snapshot: OrgSnapshot) -> None:
for child in self._sections_box.get_children():
self._sections_box.remove(child)
child.destroy()
if not snapshot.items:
self._sections_box.add(
Label(
name="org-dropdown-empty",
label="Nothing on the agenda.",
h_align="start",
)
)
self._sections_box.show_all()
return
by_state: dict[str, list[OrgItem]] = {}
for it in snapshot.items:
by_state.setdefault(it.state, []).append(it)
for state in _state_order(list(by_state.keys())):
items = by_state.get(state, [])
if not items:
continue
section = self._build_section(state, items)
self._sections_box.add(section)
self._sections_box.show_all()
def _build_section(self, state: str, items: list[OrgItem]) -> Box:
section = Box(
name="org-dropdown-section",
orientation="v",
spacing=4,
style_classes=["org-state", f"org-state-{state.lower()}"],
)
header = Box(orientation="h", spacing=6)
header.add(
Label(
name="org-dropdown-section-title",
label=state,
h_align="start",
style_classes=["org-state-label"],
)
)
header.add(
Label(
label=f"({len(items)})",
h_align="start",
style_classes=["org-state-count"],
)
)
section.add(header)
for item in items:
section.add(self._build_item(item))
return section
def _build_item(self, item: OrgItem) -> Button:
title_parts = []
if item.priority:
title_parts.append(f"#{item.priority}")
title_parts.append(item.headline)
if item.tags:
title_parts.append(":" + ":".join(item.tags) + ":")
title = " ".join(title_parts)
title_label = Label(
label=title,
h_align="start",
ellipsization="end",
style_classes=["org-item-title"],
)
meta_bits = []
if item.scheduled:
meta_bits.append(f"S: {item.scheduled}")
if item.deadline:
meta_bits.append(f"D: {item.deadline}")
meta_bits.append(os.path.basename(item.file))
meta_label = Label(
label=" · ".join(meta_bits),
h_align="start",
style_classes=["org-item-meta"],
)
content = Box(orientation="v", spacing=2, h_expand=True)
content.add(title_label)
content.add(meta_label)
button = Button(
name="org-item",
child=content,
on_clicked=lambda *_: self._on_item_clicked(item),
style_classes=["org-item"],
)
return button
def _on_item_clicked(self, item: OrgItem) -> None:
_open_in_emacs(item)
self.hide()
class OrgWidget(Button):
def __init__(self, service: OrgService, dropdown: OrgDropdown | None, **kwargs):
self._service = service
self._dropdown = dropdown
self.icon = Label(label="✓", name="org-icon")
self.label = Label("0", name="org-count")
container = Box(orientation="h", spacing=4, children=[self.icon, self.label])
super().__init__(
name="org-widget",
child=container,
on_clicked=lambda *_: self._on_clicked(),
**kwargs,
)
self._service.connect("items-changed", lambda _s, snap: self._refresh(snap))
self._refresh(self._service.snapshot)
def _on_clicked(self) -> None:
if self._dropdown is not None:
self._dropdown.toggle()
def _refresh(self, snapshot: OrgSnapshot) -> None:
counted = list(ORG.get("counted_states", ["TODO", "NEXT", "IN-PROGRESS"]))
total = snapshot.total_counted(counted)
next_count = snapshot.counts.get("NEXT", 0)
in_progress = snapshot.counts.get("IN-PROGRESS", 0)
self.label.set_text(str(total))
classes = ["org-widget"]
if total == 0:
classes.append("empty")
if next_count > 0:
classes.append("has-next")
if in_progress > 0:
classes.append("has-in-progress")
self.set_style_classes(classes)
tooltip_lines = []
for state in _state_order(list(snapshot.counts.keys())):
n = snapshot.counts.get(state, 0)
if n:
tooltip_lines.append(f"{state}: {n}")
self.set_tooltip_text("\n".join(tooltip_lines) if tooltip_lines else "No open org items")
+3 -3
View File
@@ -14,7 +14,7 @@ from fabric.widgets.overlay import Overlay
from fabric.widgets.stack import Stack
from ..widgets.circle_image import CircleImage
import sims.modules.icons as icons
from sims.services.mpris import MprisPlayer, get_mpris_player_manager
from sims.services.mpris import MprisPlayerManager, MprisPlayer
from fabric import Fabricator
# from sims.modules.cavalcade import SpectrumRender
@@ -434,7 +434,7 @@ class Player(Box):
)
self.switcher.set_stack(self.player_stack)
self.switcher.set_halign(Gtk.Align.CENTER)
self.mpris_manager = get_mpris_player_manager()
self.mpris_manager = MprisPlayerManager()
players = self.mpris_manager.players
if players:
for p in players:
@@ -623,7 +623,7 @@ class PlayerSmall(Box):
self.add(self.mpris_small)
self.mpris_manager = get_mpris_player_manager()
self.mpris_manager = MprisPlayerManager()
self.mpris_player = None
self._players = {}
self._player_handlers = {}
-19
View File
@@ -1,19 +0,0 @@
"""Shared audio service.
fabric's `Audio` wraps a libcvc (`Cvc.MixerControl`) client: constructing one
opens a PulseAudio/PipeWire connection, subscribes to every stream event and
enumerates sinks/sources. Widgets were creating one per output, so keep a
single process-wide instance instead.
"""
from fabric.audio.service import Audio
_service: "Audio | None" = None
def get_audio_service() -> "Audio":
"""Return the process-wide Audio service, creating it on first use."""
global _service
if _service is None:
_service = Audio()
return _service
-15
View File
@@ -70,18 +70,3 @@ class BatteryService(Service):
def charging(self):
"""Get current charging status"""
return self._charging
_service: "BatteryService | None" = None
def get_battery_service(update_interval: int = 10000) -> "BatteryService":
"""Return the process-wide battery service, creating it on first use.
Only the first caller's `update_interval` wins, so the bar widget (10s) and
the buddy (15s) share one psutil poll instead of running their own.
"""
global _service
if _service is None:
_service = BatteryService(update_interval=update_interval)
return _service
-169
View File
@@ -1,169 +0,0 @@
"""Mood arbitration for the bar buddy.
Fuses CPU load, battery state, time of day, mail counts and click events into a
single `mood-changed` signal. Highest-priority active mood wins.
"""
import time
from datetime import datetime
from fabric.core.service import Service, Signal
from fabric.utils import invoke_repeater
from sims.services.battery import get_battery_service
from sims.services.mpris import MprisPlayer, get_mpris_player_manager
from sims.services.system_stats import get_system_stats_service
from sims.config import BATTERY, NOTMUCH
# Highest priority first. The first mood whose state is True wins.
MOOD_PRIORITY = ["petted", "excited", "busy", "bopping", "sleepy", "idle"]
PETTED_DURATION_S = 1.5
EXCITED_DURATION_S = 5.0
BUSY_CPU_THRESHOLD = 0.80
SLEEPY_BATTERY_THRESHOLD = 20 # percent
SLEEPY_HOUR_START = 23 # 23:00 inclusive
SLEEPY_HOUR_END = 7 # exclusive
TICK_INTERVAL_MS = 1000
class BuddyService(Service):
@Signal
def mood_changed(self, mood: str) -> None:
"""Emitted when the arbitrated mood changes."""
def __init__(self, notmuch_service=None, **kwargs):
super().__init__(**kwargs)
self._mood = "idle"
self._cpu = 0.0
self._battery_pct = 100
self._battery_charging = True
self._petted_until = 0.0
self._excited_until = 0.0
self._last_unread = None
self._stats = get_system_stats_service()
self._stats.connect("stats-changed", self._on_stats)
self._battery = None
if BATTERY.get("enable", False):
self._battery = get_battery_service()
self._battery.connect("battery-changed", self._on_battery)
self._notmuch = notmuch_service
if self._notmuch is not None:
self._notmuch.connect("counts-changed", self._on_notmuch)
self._mpris_manager = get_mpris_player_manager()
self._mpris_players: dict[str, MprisPlayer] = {}
for p in self._mpris_manager.players or []:
self._attach_player(p)
self._mpris_manager.connect("player-appeared", self._on_player_appeared)
self._mpris_manager.connect("player-vanished", self._on_player_vanished)
invoke_repeater(TICK_INTERVAL_MS, self._tick)
self._recompute()
# External triggers ---------------------------------------------------
@property
def mood(self) -> str:
"""Currently arbitrated mood."""
return self._mood
def pet(self):
"""Called when the buddy is clicked."""
self._petted_until = time.monotonic() + PETTED_DURATION_S
self._recompute()
# Service callbacks ---------------------------------------------------
def _on_stats(self, _service, cpu_fraction, _mem_fraction):
self._cpu = cpu_fraction
self._recompute()
def _on_battery(self, _service, percent, charging):
self._battery_pct = percent
self._battery_charging = charging
self._recompute()
def _on_notmuch(self, _service, unread, _debt):
if self._last_unread is not None and unread > self._last_unread:
self._excited_until = time.monotonic() + EXCITED_DURATION_S
self._last_unread = unread
self._recompute()
def _attach_player(self, player):
mp = MprisPlayer(player)
self._mpris_players[mp.player_name] = mp
mp.connect("changed", self._on_player_changed)
def _on_player_appeared(self, _manager, player):
self._attach_player(player)
self._recompute()
def _on_player_vanished(self, _manager, player_name: str):
self._mpris_players.pop(player_name, None)
self._recompute()
def _on_player_changed(self, _player):
self._recompute()
def _is_anything_playing(self) -> bool:
for mp in self._mpris_players.values():
try:
if mp.playback_status == "playing":
return True
except Exception:
continue
return False
def _tick(self):
# Re-arbitrate every second so transient moods (petted/excited) and
# the clock-driven sleepy window expire correctly.
self._recompute()
return True
# Arbitration ---------------------------------------------------------
def _states(self) -> dict[str, bool]:
now = time.monotonic()
hour = datetime.now().hour
is_night = hour >= SLEEPY_HOUR_START or hour < SLEEPY_HOUR_END
battery_low = (self._battery is not None
and self._battery_pct < SLEEPY_BATTERY_THRESHOLD
and not self._battery_charging)
return {
"petted": now < self._petted_until,
"excited": now < self._excited_until,
"busy": self._cpu > BUSY_CPU_THRESHOLD,
"bopping": self._is_anything_playing(),
"sleepy": battery_low or is_night,
"idle": True,
}
def _recompute(self):
states = self._states()
for mood in MOOD_PRIORITY:
if states.get(mood, False):
if mood != self._mood:
self._mood = mood
self.mood_changed(mood)
return
_service: "BuddyService | None" = None
def get_buddy_service(notmuch_service=None) -> "BuddyService":
"""Return the process-wide buddy service, creating it on first use.
`notmuch_service` only applies to the first caller (the bar that owns the
notmuch widget). One arbitrated mood and one re-arbitration ticker are
shared by every output instead of one per bar.
"""
global _service
if _service is None:
_service = BuddyService(notmuch_service=notmuch_service)
return _service
-16
View File
@@ -287,19 +287,3 @@ class MprisPlayerManager(Service):
@Property(object, "readable")
def players(self):
return self._manager.get_property("players") # type: ignore
_service: "MprisPlayerManager | None" = None
def get_mpris_player_manager() -> "MprisPlayerManager":
"""Return the process-wide MPRIS manager, creating it on first use.
`Playerctl.PlayerManager.new()` plus one `Playerctl.Player` per MPRIS name
is the expensive part; every widget that wraps players should share this
manager and build its own `MprisPlayer` wrappers on top of it.
"""
global _service
if _service is None:
_service = MprisPlayerManager()
return _service
-172
View File
@@ -1,172 +0,0 @@
"""Org-mode agenda service.
Parses configured org files via `orgparse`, exposes a flat list of TODO-state
items grouped by state. Polls on an interval and re-reads files whose mtime
changed since the last sweep.
"""
from __future__ import annotations
import os
from dataclasses import dataclass, field
from glob import glob
from pathlib import Path
from typing import Iterable
from fabric.utils import invoke_repeater
from loguru import logger
try:
import orgparse
ORGPARSE_AVAILABLE = True
except ImportError:
ORGPARSE_AVAILABLE = False
logger.warning("[Org] orgparse not available — org widget disabled at runtime")
from sims.config import ORG
@dataclass
class OrgItem:
state: str
headline: str
file: str
line: int
tags: tuple[str, ...] = ()
scheduled: str | None = None
deadline: str | None = None
priority: str | None = None
@dataclass
class OrgSnapshot:
items: list[OrgItem] = field(default_factory=list)
counts: dict[str, int] = field(default_factory=dict)
def total_counted(self, counted_states: list[str]) -> int:
return sum(self.counts.get(s, 0) for s in counted_states)
class OrgService:
def __init__(self, update_interval: int = 60_000):
self._update_interval = update_interval
self._timer_id = None
self._mtimes: dict[str, float] = {}
self._snapshot = OrgSnapshot()
self._callbacks: list = []
self.update()
self._start_monitoring()
@property
def snapshot(self) -> OrgSnapshot:
return self._snapshot
def connect(self, signal_name: str, callback) -> None:
if signal_name == "items-changed":
self._callbacks.append(callback)
def _emit(self, snapshot: OrgSnapshot) -> None:
for cb in self._callbacks:
cb(self, snapshot)
def _start_monitoring(self):
if self._timer_id is None:
self._timer_id = invoke_repeater(self._update_interval, self._tick)
logger.info(
f"[Org] Started periodic updates every {self._update_interval/1000:.0f}s"
)
def _tick(self):
self.update()
return True
def _expand_paths(self) -> list[Path]:
raw = ORG.get("paths", []) or []
out: list[Path] = []
for entry in raw:
entry = os.path.expanduser(entry)
p = Path(entry)
if any(c in entry for c in "*?["):
for match in sorted(glob(entry, recursive=True)):
mp = Path(match)
if mp.is_file() and mp.suffix == ".org":
out.append(mp)
continue
if p.is_dir():
for match in sorted(p.rglob("*.org")):
out.append(match)
elif p.is_file():
out.append(p)
return out
def update(self):
if not ORG.get("enable", False):
return
if not ORGPARSE_AVAILABLE:
return
files = self._expand_paths()
if not files:
if self._snapshot.items:
self._snapshot = OrgSnapshot()
self._emit(self._snapshot)
return
# mtime-based change detection — skip parse if nothing changed
new_mtimes: dict[str, float] = {}
for f in files:
try:
new_mtimes[str(f)] = f.stat().st_mtime
except OSError:
continue
if new_mtimes == self._mtimes and self._snapshot.items:
return
self._mtimes = new_mtimes
todo_keywords = list(ORG.get("todo_keywords", ["TODO", "NEXT", "IN-PROGRESS"]))
done_keywords = list(ORG.get("done_keywords", ["DONE", "CANCELLED"]))
items: list[OrgItem] = []
for f in files:
try:
items.extend(_parse_file(f, todo_keywords, done_keywords))
except Exception as e:
logger.error(f"[Org] Failed to parse {f}: {e}")
counts: dict[str, int] = {}
for it in items:
counts[it.state] = counts.get(it.state, 0) + 1
self._snapshot = OrgSnapshot(items=items, counts=counts)
logger.info(f"[Org] Parsed {len(items)} items across {len(files)} files: {counts}")
self._emit(self._snapshot)
def _parse_file(
path: Path,
todo_keywords: list[str],
done_keywords: list[str],
) -> Iterable[OrgItem]:
env = orgparse.OrgEnv(
todos=todo_keywords,
dones=done_keywords,
filename=str(path),
)
root = orgparse.load(str(path), env=env)
for node in root[1:]:
state = node.todo
if not state or state in done_keywords:
continue
yield OrgItem(
state=state,
headline=node.heading or "(untitled)",
file=str(path),
line=node.linenumber,
tags=tuple(sorted(node.tags)),
scheduled=str(node.scheduled) if node.scheduled else None,
deadline=str(node.deadline) if node.deadline else None,
priority=node.priority,
)
-16
View File
@@ -64,19 +64,3 @@ class SystemStatsService(Service):
def memory_percent(self):
"""Get current memory percentage"""
return self._memory_percent / 100
_service: "SystemStatsService | None" = None
def get_system_stats_service(update_interval: int = 3000) -> "SystemStatsService":
"""Return the process-wide stats service, creating it on first use.
`update_interval` only applies to the first caller: one poller feeds every
bar, and psutil's non-blocking `cpu_percent` keeps a process-global baseline
that extra pollers would keep resetting.
"""
global _service
if _service is None:
_service = SystemStatsService(update_interval=update_interval)
return _service
-26
View File
@@ -26,18 +26,6 @@
border-radius: 4px;
}
#buddy-container,
#buddy {
padding: 0;
}
#buddy-button {
padding: 0;
margin: 0 2px;
background: transparent;
border: none;
}
#bat-icon {
color: var(--blue);
margin-right: 2px;
@@ -53,20 +41,6 @@
font-weight: bold;
}
#microphone-icon {
color: var(--foreground);
font-size: 20px;
}
#microphone-button {
min-width: 28px;
padding: 0 6px;
}
#microphone-icon.muted {
color: var(--red);
}
#cpu-progress-bar,
#ram-progress-bar,
#volume-progress-bar {
+1
View File
@@ -73,6 +73,7 @@
#notmuch-date {
font-size: 11px;
opacity: 0.6;
min-width: 90px;
}
#notmuch-saved-name {
-1
View File
@@ -7,7 +7,6 @@
@import url("./calendar.css");
@import url("./notmuch.css");
@import url("./notifications.css");
@import url("./org.css");
/* unset so we can style everything from the ground up. */
-128
View File
@@ -1,128 +0,0 @@
/* Org-agenda bar widget + dropdown */
#org-widget {
background-color: var(--module-bg);
padding: 4px 8px;
border-radius: 12px;
transition: background-color 0.2s ease;
}
#org-widget:hover {
background-color: var(--light-bg);
}
#org-widget.empty {
background-color: var(--module-bg);
opacity: 0.6;
}
#org-widget.has-next {
background-color: var(--blue);
}
#org-widget.has-next:hover {
background-color: var(--turquoise);
}
#org-widget.has-in-progress {
background-color: var(--violet);
}
#org-widget.has-in-progress.has-next {
background-color: var(--turquoise);
}
#org-count {
font-size: 14px;
font-weight: bold;
min-width: 16px;
}
#org-widget.has-next #org-count,
#org-widget.has-in-progress #org-count {
color: var(--background);
}
/* Dropdown */
#org-dropdown {
background-color: var(--window-bg);
border-radius: 0 0 16px 16px;
border: 1px solid var(--border-color);
border-top: none;
}
#org-dropdown-body {
padding: 14px 16px;
}
#org-dropdown-header {
margin-bottom: 8px;
}
#org-dropdown-title {
font-size: 18px;
font-weight: bold;
}
#org-dropdown-close {
padding: 4px;
border-radius: 8px;
transition: background-color 0.15s ease;
}
#org-dropdown-close:hover {
background-color: var(--light-bg);
}
#org-dropdown-empty {
color: var(--light-grey);
font-style: italic;
padding: 12px 4px;
}
#org-dropdown-section {
padding: 6px 0;
}
#org-dropdown-section-title {
font-size: 13px;
font-weight: bold;
letter-spacing: 0.6px;
}
.org-state-next #org-dropdown-section-title {
color: var(--blue);
}
.org-state-in-progress #org-dropdown-section-title {
color: var(--violet);
}
.org-state-todo #org-dropdown-section-title {
color: var(--orange);
}
.org-state-count {
color: var(--light-grey);
font-size: 12px;
}
#org-item {
padding: 6px 8px;
border-radius: 8px;
transition: background-color 0.15s ease;
}
#org-item:hover {
background-color: var(--light-bg);
}
.org-item-title {
font-size: 14px;
}
.org-item-meta {
color: var(--light-grey);
font-size: 11px;
}
-388
View File
@@ -1,388 +0,0 @@
from gi.repository import GdkPixbuf, GLib
from fabric.widgets.box import Box
from fabric.widgets.image import Image
from fabric.widgets.button import Button
from sims.config import STYLIX
# 16x16 sprite grid, displayed at SCALE x scale (32x32) with crisp pixels.
SPRITE_SIZE = 16
SCALE = 2
# Palette character → semantic role. Resolved to RGBA from stylix at render time.
# . transparent
# B body main
# H body highlight (lighter)
# b body shadow (darker)
# e eye dark
# w eye sparkle / white
# m mouth / closed-eye line
# c blush
# s sweat drop
# z z text
# * sparkle / excitement
# _ semi-transparent ground shadow
# ---------------------------------------------------------------------------
# Sprite construction: layered grid with strict 16x16 validation.
# ---------------------------------------------------------------------------
def _grid(rows: list[str]) -> list[str]:
if len(rows) != SPRITE_SIZE:
raise ValueError(f"sprite must be {SPRITE_SIZE} rows, got {len(rows)}")
for i, r in enumerate(rows):
if len(r) != SPRITE_SIZE:
raise ValueError(f"row {i} has {len(r)} chars (want {SPRITE_SIZE}): {r!r}")
return rows
def _overlay(base: list[str], *layers: list[tuple[int, int, str]]) -> list[str]:
grid = [list(r) for r in base]
for layer in layers:
for (y, x, ch) in layer:
if 0 <= y < SPRITE_SIZE and 0 <= x < SPRITE_SIZE:
grid[y][x] = ch
return ["".join(r) for r in grid]
def _shift(base: list[str], dy: int = 0, dx: int = 0) -> list[str]:
out = []
for y in range(SPRITE_SIZE):
src_y = y - dy
if 0 <= src_y < SPRITE_SIZE:
row = base[src_y]
else:
row = "." * SPRITE_SIZE
if dx > 0:
row = "." * dx + row[:-dx]
elif dx < 0:
row = row[-dx:] + "." * (-dx)
out.append(row)
return out
def _shift_overlay(overlay: list[tuple[int, int, str]], dy: int = 0, dx: int = 0) -> list[tuple[int, int, str]]:
return [(y + dy, x + dx, ch) for (y, x, ch) in overlay]
# Body shapes ---------------------------------------------------------------
BODY_REST = _grid([
"................",
"................",
"................",
".......BB.......",
"......BBBB......",
".....BBBBBB.....",
"....HBBBBBBB....",
"...HBBBBBBBBB...",
"..HBBBBBBBBBBB..",
"..BBBBBBBBBBBB..",
".BBBBBBBBBBBBBb.",
".BBBBBBBBBBBBBb.",
".BBBBBBBBBBBBbb.",
"BBBBBBBBBBBBBbbb",
"BBBBBBBBBBBBBbbb",
"................",
])
BODY_BOB = _shift(BODY_REST, dy=-1) # bobbed up by 1 row
BODY_SQUISH = _grid([
"................",
"................",
"................",
"................",
"......BBBB......",
".....BBBBBB.....",
"....BBBBBBBB....",
"...HBBBBBBBBB...",
"..HBBBBBBBBBBB..",
"..BBBBBBBBBBBB..",
".BBBBBBBBBBBBBb.",
".BBBBBBBBBBBBBb.",
"BBBBBBBBBBBBBbbb",
"BBBBBBBBBBBBBbbb",
"BBBBBBBBBBBBBbbb",
"................",
])
BODY_LEAN_L = _shift(BODY_REST, dx=-1)
BODY_LEAN_R = _shift(BODY_REST, dx=1)
# Face overlays. Eye row defaults to row 8 (rest pose); pass r= for bob/squish.
def eyes_open(r: int = 8) -> list[tuple[int, int, str]]:
return [(r, 5, "e"), (r, 10, "e")]
def eyes_blink(r: int = 8) -> list[tuple[int, int, str]]:
return [(r, 4, "m"), (r, 5, "m"), (r, 9, "m"), (r, 10, "m")]
def eyes_closed_happy(r: int = 8) -> list[tuple[int, int, str]]:
# ^^ shaped happy eyes
return [(r - 1, 4, "m"), (r - 1, 9, "m"),
(r, 5, "m"), (r, 10, "m"),
(r, 3, "m"), (r, 8, "m")]
def eyes_wide(r: int = 8) -> list[tuple[int, int, str]]:
return [(r - 1, 5, "w"), (r - 1, 10, "w"),
(r, 5, "e"), (r, 10, "e")]
def mouth_neutral(r: int = 11) -> list[tuple[int, int, str]]:
return [(r, 6, "m"), (r, 7, "m"), (r, 8, "m"), (r, 9, "m")]
def mouth_smile(r: int = 11) -> list[tuple[int, int, str]]:
return [(r, 6, "m"), (r, 9, "m"),
(r + 1, 7, "m"), (r + 1, 8, "m")]
def mouth_frown(r: int = 11) -> list[tuple[int, int, str]]:
return [(r, 7, "m"), (r, 8, "m"),
(r - 1, 6, "m"), (r - 1, 9, "m")]
def mouth_o(r: int = 11) -> list[tuple[int, int, str]]:
return [(r, 7, "m"), (r, 8, "m"),
(r + 1, 7, "m"), (r + 1, 8, "m")]
def blush(r: int = 9) -> list[tuple[int, int, str]]:
return [(r, 3, "c"), (r, 12, "c")]
# Accent overlays
SWEAT_HIGH = [(2, 13, "s"), (3, 13, "s")]
SWEAT_LOW = [(5, 13, "s"), (6, 13, "s")]
Z_SMALL = [(3, 12, "z"), (3, 13, "z")]
Z_LARGE = [(1, 13, "z"), (1, 14, "z"),
(2, 13, "z"), (2, 14, "z"),
(3, 12, "z"), (3, 13, "z")]
SPARKLE_LEFT = [(3, 2, "*"), (5, 1, "*")]
SPARKLE_RIGHT = [(3, 14, "*"), (5, 14, "*")]
# ---------------------------------------------------------------------------
# Frame compositions per mood
# ---------------------------------------------------------------------------
# Idle: bob + occasional blink. Most frames are eyes-open at rest.
IDLE_REST = _overlay(BODY_REST, eyes_open(8), mouth_neutral(11))
IDLE_BOB = _overlay(BODY_BOB, eyes_open(7), mouth_neutral(10))
IDLE_BLINK = _overlay(BODY_REST, eyes_blink(8), mouth_neutral(11))
# Petted: closed-happy eyes + smile + blush. Squish lowers face by 1 row.
PETTED_FLAT = _overlay(BODY_SQUISH, eyes_closed_happy(9), mouth_smile(12), blush(10))
PETTED_RECOVER = _overlay(BODY_REST, eyes_closed_happy(8), mouth_smile(11), blush(9))
# Excited: wide eyes, open mouth, sparkles, bouncing
EXCITED_LOW = _overlay(BODY_REST, eyes_wide(8), mouth_o(11),
SPARKLE_LEFT)
EXCITED_HIGH = _overlay(BODY_BOB, eyes_wide(7), mouth_o(10),
SPARKLE_LEFT, SPARKLE_RIGHT)
# Bopping: sway in time with music. Shift face with the body.
BOP_LEFT = _overlay(BODY_LEAN_L,
_shift_overlay(eyes_open(8), dx=-1),
_shift_overlay(mouth_smile(11), dx=-1))
BOP_RIGHT = _overlay(BODY_LEAN_R,
_shift_overlay(eyes_open(8), dx=1),
_shift_overlay(mouth_smile(11), dx=1))
# Busy: frown + sweat drops
BUSY_DROP_HIGH = _overlay(BODY_REST, eyes_open(8), mouth_frown(11), SWEAT_HIGH)
BUSY_DROP_LOW = _overlay(BODY_REST, eyes_open(8), mouth_frown(11), SWEAT_LOW)
# Sleepy: closed eyes, z floats, no mouth
SLEEPY_NO_Z = _overlay(BODY_REST, eyes_blink(8))
SLEEPY_Z_SMALL = _overlay(BODY_REST, eyes_blink(8), Z_SMALL)
SLEEPY_Z_LARGE = _overlay(BODY_REST, eyes_blink(8), Z_LARGE)
# (frames, frame_duration_ms, loop)
MOOD_FRAMES: dict[str, tuple[list[list[str]], int, bool]] = {
"idle": (
[IDLE_REST, IDLE_REST, IDLE_BOB, IDLE_REST, IDLE_REST, IDLE_BLINK],
300, True,
),
"petted": (
[PETTED_FLAT, PETTED_FLAT, PETTED_RECOVER, PETTED_RECOVER],
180, False,
),
"excited": (
[EXCITED_LOW, EXCITED_HIGH],
180, True,
),
"bopping": (
[BOP_LEFT, IDLE_REST, BOP_RIGHT, IDLE_REST],
220, True,
),
"busy": (
[BUSY_DROP_HIGH, BUSY_DROP_LOW],
260, True,
),
"sleepy": (
[SLEEPY_NO_Z, SLEEPY_NO_Z, SLEEPY_Z_SMALL, SLEEPY_Z_LARGE],
500, True,
),
}
# ---------------------------------------------------------------------------
# Pixbuf rendering
# ---------------------------------------------------------------------------
def _hex_to_rgba(hex_str: str, alpha: int = 255) -> tuple[int, int, int, int]:
h = hex_str.lstrip("#")
return (int(h[0:2], 16), int(h[2:4], 16), int(h[4:6], 16), alpha)
def _mix(c1: tuple[int, int, int], c2: tuple[int, int, int], t: float) -> tuple[int, int, int]:
return (int(c1[0] * (1 - t) + c2[0] * t),
int(c1[1] * (1 - t) + c2[1] * t),
int(c1[2] * (1 - t) + c2[2] * t))
def _resolve_palette() -> dict[str, tuple[int, int, int, int]]:
defaults = {
"base00": "1e1e2e", "base05": "cdd6f4", "base07": "b4befe",
"base08": "f38ba8", "base0A": "f9e2af", "base0C": "94e2d5",
"base0D": "89b4fa",
}
colors = dict(STYLIX.get("colors", {})) if STYLIX.get("enable", False) else {}
for k, v in defaults.items():
colors.setdefault(k, v)
body_rgba = _hex_to_rgba(colors["base0D"])
body_rgb = body_rgba[:3]
light_rgb = _mix(body_rgb, (255, 255, 255), 0.35)
dark_rgb = _mix(body_rgb, (0, 0, 0), 0.30)
return {
".": (0, 0, 0, 0),
"B": body_rgba,
"H": light_rgb + (255,),
"b": dark_rgb + (255,),
"e": _hex_to_rgba(colors["base00"]),
"w": _hex_to_rgba(colors["base05"]),
"m": _hex_to_rgba(colors["base00"]),
"c": _hex_to_rgba(colors["base08"]),
"s": _hex_to_rgba(colors["base0C"]),
"z": _hex_to_rgba(colors["base05"]),
"*": _hex_to_rgba(colors["base0A"]),
"_": _hex_to_rgba(colors["base00"], alpha=80),
}
def _frame_to_pixbuf(rows: list[str], palette: dict[str, tuple[int, int, int, int]]) -> GdkPixbuf.Pixbuf:
out_size = SPRITE_SIZE * SCALE
rowstride = out_size * 4
buf = bytearray(rowstride * out_size)
for y in range(SPRITE_SIZE):
for x in range(SPRITE_SIZE):
ch = rows[y][x]
r, g, b, a = palette.get(ch, (0, 0, 0, 0))
for sy in range(SCALE):
for sx in range(SCALE):
o = (y * SCALE + sy) * rowstride + (x * SCALE + sx) * 4
buf[o] = r
buf[o + 1] = g
buf[o + 2] = b
buf[o + 3] = a
return GdkPixbuf.Pixbuf.new_from_bytes(
GLib.Bytes.new(bytes(buf)),
GdkPixbuf.Colorspace.RGB,
True, 8,
out_size, out_size, rowstride,
)
# ---------------------------------------------------------------------------
# Widget
# ---------------------------------------------------------------------------
class BuddyWidget(Box):
def __init__(self, on_pet=None, **kwargs):
super().__init__(name="buddy", v_align="center", **kwargs)
self._palette = _resolve_palette()
self._cache: dict[int, GdkPixbuf.Pixbuf] = {}
self._mood = "idle"
self._frame_idx = 0
self._timer_id = None
self._on_pet = on_pet
out = SPRITE_SIZE * SCALE
self._image = Image(name="buddy-image")
self._image.set_size_request(out, out)
self._button = Button(
name="buddy-button",
child=self._image,
on_clicked=self._handle_click,
style="background: transparent; border: none; padding: 0; margin: 0; box-shadow: none;",
)
self.children = [self._button]
self._set_mood("idle")
self.show_all()
def _pixbuf_for(self, frame_rows: list[str]) -> GdkPixbuf.Pixbuf:
key = id(frame_rows)
if key not in self._cache:
self._cache[key] = _frame_to_pixbuf(frame_rows, self._palette)
return self._cache[key]
def _set_mood(self, mood: str):
if mood not in MOOD_FRAMES:
mood = "idle"
self._mood = mood
self._frame_idx = 0
if self._timer_id is not None:
GLib.source_remove(self._timer_id)
self._timer_id = None
_frames, dur, _loop = MOOD_FRAMES[mood]
self._render_current()
self._timer_id = GLib.timeout_add(dur, self._tick)
def _tick(self):
frames, _dur, loop = MOOD_FRAMES[self._mood]
self._frame_idx += 1
if self._frame_idx >= len(frames):
if loop:
self._frame_idx = 0
else:
self._frame_idx = len(frames) - 1
self._timer_id = None
if callable(self._on_pet) and self._mood == "petted":
GLib.timeout_add(120, self._notify_petted_done)
return False
self._render_current()
return True
def _render_current(self):
frames, _dur, _loop = MOOD_FRAMES[self._mood]
rows = frames[self._frame_idx]
self._image.set_from_pixbuf(self._pixbuf_for(rows))
def _notify_petted_done(self):
if callable(self._on_pet):
self._on_pet(done=True)
return False
def _handle_click(self, *_args):
if callable(self._on_pet):
self._on_pet(petted=True)
def update_mood(self, mood: str):
if mood == self._mood:
return
self._set_mood(mood)