diff --git a/pomodoro_app/packaging/arch/pomodoro-wake-daemon.py b/pomodoro_app/packaging/arch/pomodoro-wake-daemon.py index 8f26365..3a5686d 100755 --- a/pomodoro_app/packaging/arch/pomodoro-wake-daemon.py +++ b/pomodoro_app/packaging/arch/pomodoro-wake-daemon.py @@ -36,6 +36,9 @@ logging.basicConfig( log = logging.getLogger(__name__) +MIN_DEVICE_PARTS = 2 + + def is_app_running() -> bool: """Check whether the Pomodoro app is running locally.""" pgrep = shutil.which("pgrep") @@ -88,7 +91,7 @@ def get_adb_devices() -> list[str]: devices: list[str] = [] for line in result.stdout.strip().splitlines()[1:]: parts = line.split() - if len(parts) >= 2 and parts[1] == "device": # noqa: PLR2004 + if len(parts) >= MIN_DEVICE_PARTS and parts[1] == "device": devices.append(parts[0]) return devices diff --git a/python_pkg/download_cats/http_status_anki.py b/python_pkg/download_cats/http_status_anki.py index 07a27b5..0e59b36 100755 --- a/python_pkg/download_cats/http_status_anki.py +++ b/python_pkg/download_cats/http_status_anki.py @@ -13,6 +13,7 @@ import hashlib import logging from pathlib import Path import sys +import tempfile from typing import TYPE_CHECKING import genanki @@ -357,7 +358,7 @@ class _DeckBuilder: filename = f"http_cat_{status_code}.jpg" # Save to temp directory for genanki - temp_path = Path(f"/tmp/{filename}") # noqa: S108 + temp_path = Path(tempfile.gettempdir()) / filename temp_path.write_bytes(image_data) self.media_files.append(str(temp_path)) diff --git a/python_pkg/music_gen/music_generator.py b/python_pkg/music_gen/music_generator.py index 3bcb84c..34ad8ec 100755 --- a/python_pkg/music_gen/music_generator.py +++ b/python_pkg/music_gen/music_generator.py @@ -39,28 +39,21 @@ def check_dependencies(*, include_bark: bool = False) -> bool: Args: include_bark: Whether to check for Bark dependencies as well. """ + import importlib.util + missing = [] - try: - import torch # noqa: F401 - except ImportError: + if importlib.util.find_spec("torch") is None: missing.append("torch") - try: - import transformers # noqa: F401 - except ImportError: + if importlib.util.find_spec("transformers") is None: missing.append("transformers") - try: - import scipy # noqa: F401 - except ImportError: + if importlib.util.find_spec("scipy") is None: missing.append("scipy") - if include_bark: - try: - from bark import generate_audio as _bark_gen # noqa: F401 - except ImportError: - missing.append("git+https://github.com/suno-ai/bark.git") + if include_bark and importlib.util.find_spec("bark") is None: + missing.append("git+https://github.com/suno-ai/bark.git") if missing: print("Missing dependencies. Install with:") diff --git a/python_pkg/repo_explorer/repo_explorer.py b/python_pkg/repo_explorer/repo_explorer.py index c98e0d2..9f0a2b3 100755 --- a/python_pkg/repo_explorer/repo_explorer.py +++ b/python_pkg/repo_explorer/repo_explorer.py @@ -469,7 +469,7 @@ class RepoExplorer(tk.Tk): fcntl.fcntl(master_fd, fcntl.F_SETFL, fl | os.O_NONBLOCK) self._proc = subprocess.Popen( - ["bash", "run.sh", *extra], # noqa: S607 + ["/usr/bin/bash", "run.sh", *extra], cwd=path, stdin=slave_fd, stdout=slave_fd, @@ -485,7 +485,27 @@ class RepoExplorer(tk.Tk): threading.Thread(target=self._read_pty, daemon=True).start() threading.Thread(target=self._wait_proc, daemon=True).start() - def _read_pty(self) -> None: # noqa: C901, PLR0912 + @staticmethod + def _decode_buf(buf: bytes) -> str: + """Decode a byte buffer, strip ANSI codes and carriage returns.""" + return _strip_ansi(buf.decode("utf-8", errors="replace").replace("\r", "")) + + def _flush_partial_buf(self, buf: bytes) -> None: + """Flush a partial (no trailing newline) buffer to output.""" + text = self._decode_buf(buf) + if text: + self._write_output(text) + + def _process_complete_lines(self, buf: bytes) -> bytes: + """Split buf on newlines, output complete lines, return remainder.""" + while b"\n" in buf: + line, buf = buf.split(b"\n", 1) + text = self._decode_buf(line) + if text: + self._write_output(text + "\n") + return buf + + def _read_pty(self) -> None: """Stream PTY output to the widget, stripping ANSI codes. Partial lines (prompts without a trailing newline) are flushed after @@ -503,11 +523,7 @@ class RepoExplorer(tk.Tk): if buf: idle_ticks += 1 if idle_ticks >= self._IDLE_FLUSH_TICKS: - text = _strip_ansi( - buf.decode("utf-8", errors="replace").replace("\r", "") - ) - if text: - self._write_output(text) + self._flush_partial_buf(buf) buf = b"" idle_ticks = 0 continue @@ -519,18 +535,10 @@ class RepoExplorer(tk.Tk): if not chunk: break buf += chunk - while b"\n" in buf: - line, buf = buf.split(b"\n", 1) - text = _strip_ansi( - line.decode("utf-8", errors="replace").replace("\r", "") - ) - if text: - self._write_output(text + "\n") + buf = self._process_complete_lines(buf) # flush remainder if buf: - text = _strip_ansi(buf.decode("utf-8", errors="replace").replace("\r", "")) - if text: - self._write_output(text) + self._flush_partial_buf(buf) if self._master_fd is not None: with contextlib.suppress(OSError): os.close(self._master_fd)