From 6a1e8c592b3367b1cda610d0af5f5f260ae5e9bf Mon Sep 17 00:00:00 2001 From: Nick Marcel <17437302+nyannoying1337@users.noreply.github.com> Date: Fri, 11 Sep 2026 02:33:43 +0200 Subject: [PATCH 1/3] Add pause_when so FetchCord stops overriding other statuses Discord shows one activity at a time, so while a game or Spotify is running FetchCord's system info replaces the status the user actually wants on their profile. There's currently no way to say "let that one win". pause_when lists the programs that should. While any of them is running the presence is cleared and the loop just waits; when they exit it picks up where it left off. The .exe suffix is optional in both directions, so one entry works on every platform. Configurable two ways: fetchcord --pause-when steam spotify or pause_when in fetchcord_conf.yml. The CLI wins. I added the flag because Config() only ever reads the packaged YAML - there's no %APPDATA%/XDG lookup - so a config-only version of this would mean editing files inside site-packages to use it. Matching is on the whole process name, not a substring, so "steam" doesn't get triggered by steamwebhelper. The transitions are announced once each rather than every cycle. psutil is already a dependency, so nothing new is pulled in. --- fetch_cord/__main__.py | 17 ++++ fetch_cord/args.py | 10 ++ fetch_cord/processes.py | 90 ++++++++++++++++++ fetch_cord/resources/fetchcord_conf.yml | 9 ++ tests/test_more.py | 101 ++++++++++++++++++++ tests/test_processes.py | 121 ++++++++++++++++++++++++ 6 files changed, 348 insertions(+) create mode 100644 fetch_cord/processes.py create mode 100644 tests/test_processes.py diff --git a/fetch_cord/__main__.py b/fetch_cord/__main__.py index 0cc2736..e869f94 100755 --- a/fetch_cord/__main__.py +++ b/fetch_cord/__main__.py @@ -13,6 +13,7 @@ from fetch_cord.config import Config from fetch_cord.constants import ( CUSTOM_TIME_MESSAGE, + DEFAULT_CYCLE_TIME_SECONDS, MIN_CYCLE_TIME_SECONDS, RESULT_NOT_FOUND, ) @@ -25,6 +26,7 @@ get_component_id, get_infos, ) +from fetch_cord.processes import PauseWatcher from fetch_cord.resources import systemd_service from fetch_cord.update import update @@ -110,6 +112,11 @@ def main( if args.time: cycle.time = int(args.time) + # CLI wins over the config file so this is usable without editing the + # packaged config. + pause_when = getattr(args, "pause_when", None) or config.get("pause_when") or [] + pause = PauseWatcher(pause_when) + os_type = platform.system() # Only the commands defined for this OS are run. The structured fastfetch # provider fills the common fields first; command/native providers only @@ -151,6 +158,16 @@ def signal_handler(signum: int, frame: object) -> None: ): continue + if pause.check(): + # Leave the profile alone so whatever else is running keeps + # the status it would have had. + for other in cycles: + if other.rpc: + other.close_connection() + current_client_id = None + cycle.wait(cycle.time or DEFAULT_CYCLE_TIME_SECONDS) + continue + # Collect every field once per cycle, then read from the snapshot. snapshot = fetch.snapshot() app = snapshot.get(app_id, RESULT_NOT_FOUND) diff --git a/fetch_cord/args.py b/fetch_cord/args.py index 22653d8..66f3c39 100755 --- a/fetch_cord/args.py +++ b/fetch_cord/args.py @@ -67,6 +67,16 @@ def parse_args() -> argparse.Namespace: action="store_true", help="Get files from testing branch instead of master.", ) + parser.add_argument( + "--pause-when", + metavar="NAME", + nargs="+", + help=( + "Process names that should hold the presence back while they run " + "(e.g. --pause-when steam spotify). Overrides pause_when in the " + "config. The .exe suffix is optional." + ), + ) parser.add_argument("--debug", "-d", action="store_true", help="Enable debugging.") parser.add_argument( "--version", "-v", action="store_true", help="Print FetchCord Version." diff --git a/fetch_cord/processes.py b/fetch_cord/processes.py new file mode 100644 index 0000000..44dc86e --- /dev/null +++ b/fetch_cord/processes.py @@ -0,0 +1,90 @@ +"""Noticing when another program should own the Rich Presence instead. + +Discord shows one activity at a time, so while a game or Spotify is running +FetchCord's system info replaces the status the user actually wants to see. +``pause_when`` lists the programs that should win. +""" + +from collections.abc import Iterable + +import psutil + + +def _variants(name: str) -> set[str]: + """A process name with and without its ``.exe`` suffix. + + Lets one config line ("steam") match on every platform. + """ + name = name.strip().lower() + if not name: + return set() + + if name.endswith(".exe"): + return {name, name[: -len(".exe")]} + return {name, name + ".exe"} + + +def running_names() -> list[str]: + """Lowercased names of every process we can see.""" + names: list[str] = [] + + try: + processes = psutil.process_iter(["name"]) + except psutil.Error: + return [] + + for process in processes: + try: + name = process.info.get("name") + except (psutil.Error, KeyError): + # A process exiting between listing and reading is normal and + # must not abandon the whole scan. + continue + if name: + names.append(name.lower()) + + return names + + +def find_running(names: Iterable[str]) -> str | None: + """The first of ``names`` currently running, as the caller spelled it.""" + wanted: dict[str, str] = {} + for name in names: + for variant in _variants(name): + wanted[variant] = name.strip() + + if not wanted: + return None + + for running in running_names(): + if running in wanted: + return wanted[running] + + return None + + +class PauseWatcher: + """Tracks which configured program is holding the presence back. + + Announces the transitions only. This is checked once per cycle, so a line + every time would be noise. + """ + + def __init__(self, names: Iterable[str] | None = None) -> None: + self.names = [name for name in (names or []) if name and name.strip()] + self.paused_for: str | None = None + + def check(self) -> str | None: + """The program currently pausing us, or None.""" + if not self.names: + return None + + found = find_running(self.names) + + if found and found != self.paused_for: + print(f"Pausing: {found} is running.") + elif self.paused_for and not found: + print(f"{self.paused_for} closed, resuming.") + + self.paused_for = found + return found diff --git a/fetch_cord/resources/fetchcord_conf.yml b/fetch_cord/resources/fetchcord_conf.yml index 28bc54b..9da9765 100644 --- a/fetch_cord/resources/fetchcord_conf.yml +++ b/fetch_cord/resources/fetchcord_conf.yml @@ -1,4 +1,13 @@ --- +# Programs that should keep their own Discord status while they run. +# FetchCord clears its presence until they exit. The .exe suffix is +# optional, so one entry works on every platform. +# +# pause_when: +# - steam +# - spotify +pause_when: [] + cycle_time: &cycle_time 30 cycles: - name: os diff --git a/tests/test_more.py b/tests/test_more.py index 9b4f020..4929ed6 100644 --- a/tests/test_more.py +++ b/tests/test_more.py @@ -413,6 +413,107 @@ def test_main_loop( fake_fetch.snapshot.assert_called() +class TestPauseWhenInTheLoop(unittest.TestCase): + def _ns(self, **kwargs: object) -> argparse.Namespace: + defaults: dict[str, object] = { + "update": False, + "testing": False, + "install": False, + "uninstall": False, + "enable": False, + "disable": False, + "start": False, + "stop": False, + "status": False, + "version": False, + "time": None, + "debug": False, + "pause_when": ["steam"], + "nodistro": False, + "nohardware": False, + "noshell": False, + "nohost": False, + } + defaults.update(kwargs) + return argparse.Namespace(**defaults) + + @patch("fetch_cord.processes.running_names", return_value=["steam.exe"]) + @patch("fetch_cord.__main__.Fetch") + @patch("fetch_cord.__main__.get_infos", return_value={}) + @patch("fetch_cord.cycle.Cycle.close_connection") + @patch("fetch_cord.cycle.Cycle.update") + @patch("fetch_cord.cycle.Cycle.setup") + def test_a_paused_program_stops_us_updating( + self, + mock_setup: MagicMock, + mock_update: MagicMock, + mock_close: MagicMock, + mock_get_infos: MagicMock, + mock_fetch_class: MagicMock, + mock_names: MagicMock, + ) -> None: + """The whole point: leave the profile to whatever else is running.""" + from threading import Event + + from fetch_cord.__main__ import main + + fake_fetch = MagicMock() + fake_fetch.snapshot.return_value = {} + mock_fetch_class.return_value = fake_fetch + + stop = Event() + waits: list[object] = [] + + def record_wait(*args: object, **kwargs: object) -> None: + waits.append(args) + if len(waits) >= 4: + stop.set() + + with patch("fetch_cord.cycle.Cycle.wait", side_effect=record_wait): + main(self._ns(), stop_event=stop) + + mock_update.assert_not_called() + mock_setup.assert_not_called() + # It still paces itself rather than spinning. + self.assertGreaterEqual(len(waits), 4) + + @patch("fetch_cord.processes.running_names", return_value=["firefox"]) + @patch("fetch_cord.__main__.Fetch") + @patch("fetch_cord.__main__.get_component_id", return_value="client-1") + @patch("fetch_cord.__main__.get_infos", return_value={}) + @patch("fetch_cord.cycle.Cycle.try_connect") + @patch("fetch_cord.cycle.Cycle.setup") + def test_nothing_matching_leaves_the_loop_alone( + self, + mock_setup: MagicMock, + mock_try_connect: MagicMock, + mock_get_infos: MagicMock, + mock_get_component_id: MagicMock, + mock_fetch_class: MagicMock, + mock_names: MagicMock, + ) -> None: + from threading import Event + + from fetch_cord.__main__ import main + + fake_fetch = MagicMock() + fake_fetch.snapshot.return_value = {} + mock_fetch_class.return_value = fake_fetch + + stop = Event() + updates: list[object] = [] + + def record_update(*args: object, **kwargs: object) -> None: + updates.append(args) + if len(updates) >= 2: + stop.set() + + with patch("fetch_cord.cycle.Cycle.update", side_effect=record_update): + main(self._ns(), stop_event=stop) + + self.assertGreaterEqual(len(updates), 2) + + class TestCycleExtras(unittest.TestCase): def _cycle(self: Any) -> Any: from threading import Event diff --git a/tests/test_processes.py b/tests/test_processes.py new file mode 100644 index 0000000..d7fe2f8 --- /dev/null +++ b/tests/test_processes.py @@ -0,0 +1,121 @@ +"""Tests for pausing the presence while another program owns the status. + +Discord shows one activity at a time, so these cover the decision about when +FetchCord should get out of the way - no real process scanning required. +""" + +import unittest +from unittest.mock import MagicMock, patch + +import psutil + +from fetch_cord.processes import PauseWatcher, find_running, running_names + + +def fake_processes(*names: str) -> list[MagicMock]: + return [MagicMock(info={"name": name}) for name in names] + + +class TestRunningNames(unittest.TestCase): + @patch("fetch_cord.processes.psutil.process_iter") + def test_lowercases_every_name(self, mock_iter: MagicMock) -> None: + mock_iter.return_value = fake_processes("Steam.exe", "Discord") + + self.assertEqual(running_names(), ["steam.exe", "discord"]) + + @patch("fetch_cord.processes.psutil.process_iter") + def test_skips_processes_without_a_name(self, mock_iter: MagicMock) -> None: + mock_iter.return_value = fake_processes("steam", "") + [ + MagicMock(info={"name": None}) + ] + + self.assertEqual(running_names(), ["steam"]) + + @patch("fetch_cord.processes.psutil.process_iter") + def test_a_process_exiting_mid_scan_does_not_abandon_it( + self, mock_iter: MagicMock + ) -> None: + """Processes come and go while we iterate; that's normal, not fatal.""" + dying = MagicMock() + type(dying).info = property( + lambda self: (_ for _ in ()).throw(psutil.NoSuchProcess(1)) + ) + mock_iter.return_value = [dying, *fake_processes("spotify")] + + self.assertEqual(running_names(), ["spotify"]) + + @patch( + "fetch_cord.processes.psutil.process_iter", side_effect=psutil.AccessDenied() + ) + def test_returns_empty_when_the_scan_is_refused(self, mock_iter: MagicMock) -> None: + self.assertEqual(running_names(), []) + + +class TestFindRunning(unittest.TestCase): + @patch("fetch_cord.processes.running_names") + def test_matches_with_or_without_the_exe_suffix( + self, mock_names: MagicMock + ) -> None: + """One config line has to work on Windows and on Linux.""" + mock_names.return_value = ["steam.exe"] + self.assertEqual(find_running(["steam"]), "steam") + + mock_names.return_value = ["steam"] + self.assertEqual(find_running(["steam.exe"]), "steam.exe") + + @patch("fetch_cord.processes.running_names", return_value=["discord", "spotify"]) + def test_returns_the_name_as_written(self, mock_names: MagicMock) -> None: + self.assertEqual(find_running(["Steam", "SPOTIFY"]), "SPOTIFY") + + @patch("fetch_cord.processes.running_names", return_value=["discord"]) + def test_returns_none_when_nothing_matches(self, mock_names: MagicMock) -> None: + self.assertIsNone(find_running(["steam", "spotify"])) + + @patch("fetch_cord.processes.running_names") + def test_empty_config_never_scans(self, mock_names: MagicMock) -> None: + self.assertIsNone(find_running([])) + self.assertIsNone(find_running(["", " "])) + mock_names.assert_not_called() + + @patch("fetch_cord.processes.running_names", return_value=["steam.exe"]) + def test_substrings_do_not_match(self, mock_names: MagicMock) -> None: + """ "steam" must not be matched by "steamwebhelper".""" + self.assertIsNone(find_running(["team"])) + + +class TestPauseWatcher(unittest.TestCase): + @patch("fetch_cord.processes.running_names", return_value=["steam.exe"]) + @patch("builtins.print") + def test_announces_start_and_stop_once_each( + self, mock_print: MagicMock, mock_names: MagicMock + ) -> None: + """Checked every cycle, so a line per check would be noise.""" + watcher = PauseWatcher(["steam"]) + + self.assertEqual(watcher.check(), "steam") + self.assertEqual(watcher.check(), "steam") + + mock_names.return_value = [] + self.assertIsNone(watcher.check()) + self.assertIsNone(watcher.check()) + + printed = [str(call) for call in mock_print.call_args_list] + self.assertEqual(len([p for p in printed if "Pausing" in p]), 1) + self.assertEqual(len([p for p in printed if "resuming" in p]), 1) + + @patch("fetch_cord.processes.running_names") + def test_no_names_configured_is_a_no_op(self, mock_names: MagicMock) -> None: + watcher = PauseWatcher([]) + + self.assertIsNone(watcher.check()) + mock_names.assert_not_called() + + def test_blank_entries_are_dropped(self) -> None: + self.assertEqual(PauseWatcher(["steam", "", " "]).names, ["steam"]) + + def test_none_is_accepted(self) -> None: + self.assertEqual(PauseWatcher(None).names, []) + + +if __name__ == "__main__": + unittest.main(verbosity=2) From c09eacb798f8da2ac9ceb7781244f07a914c3561 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 11 Sep 2026 17:47:45 +0000 Subject: [PATCH 2/3] Fix ruff format in test_more after merge Co-authored-by: BlivionIaG --- tests/test_more.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_more.py b/tests/test_more.py index 14076b4..39e692d 100644 --- a/tests/test_more.py +++ b/tests/test_more.py @@ -580,6 +580,7 @@ def record_update(*args: object, **kwargs: object) -> None: self.assertGreaterEqual(len(updates), 2) + class TestDryRun(unittest.TestCase): def _ns(self, **kwargs: object) -> argparse.Namespace: defaults: dict[str, object] = { @@ -675,7 +676,6 @@ def test_print_dry_run_renders_every_cycle( self.assertIn("Debian", printed) - class TestCycleExtras(unittest.TestCase): def _cycle(self: Any) -> Any: from threading import Event From 9943d13d7c2501d8defff8b4f1362fdb9efdc50e Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 11 Sep 2026 17:50:20 +0000 Subject: [PATCH 3/3] Trigger CI for pause_when follow-up PR Co-authored-by: BlivionIaG