mirror of
https://github.com/kuhyx/testsAndMisc-archive.git
synced 2026-07-04 13:23:01 +02:00
refactor: remove noqa comments from miscellaneous scripts
- Fix underlying lint issues instead of suppressing with noqa - Files: moviepy_showcase, pomodoro-wake-daemon, brother_printer, http_status_anki, geo_data, repo_explorer, steam_backlog_enforcer, music_generator
This commit is contained in:
parent
2a61619001
commit
9482719608
@ -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
|
||||
|
||||
|
||||
@ -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))
|
||||
|
||||
|
||||
@ -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:")
|
||||
|
||||
@ -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)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user