feat: launcher recommendation
This commit is contained in:
@@ -1,6 +1,7 @@
|
|||||||
import re
|
import re
|
||||||
import shlex
|
import shlex
|
||||||
import subprocess
|
import subprocess
|
||||||
|
import time
|
||||||
|
|
||||||
from fabric.utils.helpers import DesktopApp, get_desktop_applications
|
from fabric.utils.helpers import DesktopApp, get_desktop_applications
|
||||||
from fabric.widgets.box import Box
|
from fabric.widgets.box import Box
|
||||||
@@ -9,6 +10,8 @@ from fabric.widgets.label import Label
|
|||||||
from gi.repository import Gtk
|
from gi.repository import Gtk
|
||||||
|
|
||||||
from .base import FuzzyMenu
|
from .base import FuzzyMenu
|
||||||
|
from .history import LaunchHistory
|
||||||
|
from .ranking import rank_items
|
||||||
|
|
||||||
|
|
||||||
ICON_SIZE = 32
|
ICON_SIZE = 32
|
||||||
@@ -17,14 +20,27 @@ _FIELD_CODE_RE = re.compile(r"^%[fFuUickdDnNvm]$")
|
|||||||
|
|
||||||
|
|
||||||
class AppProvider:
|
class AppProvider:
|
||||||
|
def __init__(self, history: LaunchHistory | None = None):
|
||||||
|
self._history = history if history is not None else LaunchHistory()
|
||||||
|
|
||||||
def items(self) -> list[DesktopApp]:
|
def items(self) -> list[DesktopApp]:
|
||||||
return get_desktop_applications()
|
return get_desktop_applications()
|
||||||
|
|
||||||
def filter(self, items: list[DesktopApp], query: str) -> list[DesktopApp]:
|
def filter(self, items: list[DesktopApp], query: str) -> list[DesktopApp]:
|
||||||
if not query:
|
"""Rank apps against *query*, best match first.
|
||||||
return items
|
|
||||||
q = query.lower()
|
With an empty query everything matches equally, so the order is
|
||||||
return [a for a in items if _matches(a, q)]
|
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,
|
||||||
|
)
|
||||||
|
|
||||||
def render(self, item: DesktopApp) -> Gtk.Widget:
|
def render(self, item: DesktopApp) -> Gtk.Widget:
|
||||||
children: list[Gtk.Widget] = []
|
children: list[Gtk.Widget] = []
|
||||||
@@ -45,6 +61,7 @@ class AppProvider:
|
|||||||
return Box(name="slot-box", orientation="h", spacing=10, children=children)
|
return Box(name="slot-box", orientation="h", spacing=10, children=children)
|
||||||
|
|
||||||
def activate(self, item: DesktopApp) -> None:
|
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
|
# Launch in a transient systemd --user scope so the app gets its own
|
||||||
# cgroup instead of inheriting sims.service's. start_new_session alone
|
# cgroup instead of inheriting sims.service's. start_new_session alone
|
||||||
# only changes POSIX session/pgid; systemd tracks units by cgroup and
|
# only changes POSIX session/pgid; systemd tracks units by cgroup and
|
||||||
@@ -70,11 +87,36 @@ class AppProvider:
|
|||||||
item.launch()
|
item.launch()
|
||||||
|
|
||||||
|
|
||||||
def _matches(app: DesktopApp, q: str) -> bool:
|
def _fields(app: DesktopApp) -> dict[str, str | None]:
|
||||||
for field in (app.name, app.display_name, app.generic_name, app.executable):
|
return {
|
||||||
if field and q in field.lower():
|
"name": app.name,
|
||||||
return True
|
"display_name": app.display_name,
|
||||||
return False
|
"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 AppLauncher(monitor: int = 0) -> FuzzyMenu:
|
def AppLauncher(monitor: int = 0) -> FuzzyMenu:
|
||||||
|
|||||||
@@ -9,6 +9,8 @@ from gi.repository import Gdk, Gtk
|
|||||||
|
|
||||||
from sims.services.fenster import focused_output_index
|
from sims.services.fenster import focused_output_index
|
||||||
|
|
||||||
|
from .ranking import rank_items
|
||||||
|
|
||||||
|
|
||||||
class LauncherProvider(Protocol):
|
class LauncherProvider(Protocol):
|
||||||
def items(self) -> list[Any]: ...
|
def items(self) -> list[Any]: ...
|
||||||
@@ -47,10 +49,15 @@ class StaticActionProvider:
|
|||||||
return list(self._static or [])
|
return list(self._static or [])
|
||||||
|
|
||||||
def filter(self, items: list[StaticAction], query: str) -> list[StaticAction]:
|
def filter(self, items: list[StaticAction], query: str) -> list[StaticAction]:
|
||||||
if not query:
|
# Fixed menus (screenshot, power, screenrec) have no history: an
|
||||||
return items
|
# empty query keeps the declaration order, a real query is ranked
|
||||||
q = query.lower()
|
# like everything else in the launcher.
|
||||||
return [i for i in items if q in i.label.lower()]
|
return rank_items(
|
||||||
|
items,
|
||||||
|
query,
|
||||||
|
lambda action: {"label": action.label},
|
||||||
|
weights={"label": 1.0},
|
||||||
|
)
|
||||||
|
|
||||||
def render(self, item: StaticAction) -> Gtk.Widget:
|
def render(self, item: StaticAction) -> Gtk.Widget:
|
||||||
return Box(
|
return Box(
|
||||||
|
|||||||
@@ -6,6 +6,7 @@ from fabric.widgets.label import Label
|
|||||||
from gi.repository import Gtk
|
from gi.repository import Gtk
|
||||||
|
|
||||||
from .base import FuzzyMenu
|
from .base import FuzzyMenu
|
||||||
|
from .ranking import rank_items
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
@@ -34,10 +35,14 @@ class ClipboardProvider:
|
|||||||
return entries
|
return entries
|
||||||
|
|
||||||
def filter(self, items: list[ClipEntry], query: str) -> list[ClipEntry]:
|
def filter(self, items: list[ClipEntry], query: str) -> list[ClipEntry]:
|
||||||
if not query:
|
# Empty query keeps cliphist's newest-first order; a query is
|
||||||
return items
|
# ranked like every other launcher menu.
|
||||||
q = query.lower()
|
return rank_items(
|
||||||
return [e for e in items if q in e.preview.lower()]
|
items,
|
||||||
|
query,
|
||||||
|
lambda entry: {"preview": entry.preview},
|
||||||
|
weights={"preview": 1.0},
|
||||||
|
)
|
||||||
|
|
||||||
def render(self, item: ClipEntry) -> Gtk.Widget:
|
def render(self, item: ClipEntry) -> Gtk.Widget:
|
||||||
return Box(
|
return Box(
|
||||||
|
|||||||
@@ -0,0 +1,167 @@
|
|||||||
|
"""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
|
||||||
@@ -0,0 +1,263 @@
|
|||||||
|
"""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)
|
||||||
@@ -3,6 +3,15 @@ from fabric.widgets.box import Box
|
|||||||
from fabric.widgets.label import Label
|
from fabric.widgets.label import Label
|
||||||
from gi.repository import Gtk
|
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:
|
class WindowProvider:
|
||||||
def items(self) -> list[dict]:
|
def items(self) -> list[dict]:
|
||||||
@@ -34,14 +43,17 @@ class WindowProvider:
|
|||||||
return windows
|
return windows
|
||||||
|
|
||||||
def filter(self, items: list[dict], query: str) -> list[dict]:
|
def filter(self, items: list[dict], query: str) -> list[dict]:
|
||||||
if not query:
|
# No history for windows - empty query keeps workspace order,
|
||||||
return items
|
# typed queries are ranked by title/app id match quality.
|
||||||
q = query.lower()
|
return rank_items(
|
||||||
return [
|
items,
|
||||||
w for w in items
|
query,
|
||||||
if q in w.get("title", "").lower()
|
lambda window: {
|
||||||
or q in w.get("app_id", "").lower()
|
"title": window.get("title", ""),
|
||||||
]
|
"app_id": window.get("app_id", ""),
|
||||||
|
},
|
||||||
|
weights=_WINDOW_WEIGHTS,
|
||||||
|
)
|
||||||
|
|
||||||
def render(self, item: dict) -> Gtk.Widget:
|
def render(self, item: dict) -> Gtk.Widget:
|
||||||
title = item.get("title", "")
|
title = item.get("title", "")
|
||||||
|
|||||||
Reference in New Issue
Block a user