mirror of
https://github.com/kuhyx/testsAndMisc.git
synced 2026-07-04 12:43:12 +02:00
Krzysztof kuhy Rudnicki
038e08d2be
Split diet_guard/_gatelock.py, wake_alarm/_alarm.py, and the usage_report.py/_usage_report_parsing.py pair into focused sub-modules so every Python file is <= 500 lines, satisfying test_file_length.py. Install python-kasa into .venv (declared in requirements but missing after the 3.13->3.14 venv upgrade), fixing 8 failing smart_plug tests and restoring 100% coverage. Also includes prior in-progress work from the working tree: the wake_alarm Progress/View/Hardware field-grouping refactor, brother_printer query module + tests, diet_guard foodbank/state/cli updates, new shared coerce/logging_setup helpers, morning_routine orchestrator tweaks, dwm window-manager config, gaming scripts, and misc maintenance/digital-wellbeing script updates. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
64 lines
2.2 KiB
Python
64 lines
2.2 KiB
Python
"""Shared subprocess and CUPS-query helpers for the brother_printer package.
|
|
|
|
Centralises the short, non-checking command invocation and the ``lpstat``-based
|
|
USB-info parsing that the CUPS, USB, and status modules all repeat, so the
|
|
subprocess boilerplate and URI parsing live in exactly one place.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import subprocess
|
|
import urllib.parse
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def run_command_text(args: list[str], *, timeout: float = 5) -> str:
|
|
"""Run ``args`` and return its captured stdout, or "" on any failure.
|
|
|
|
A non-checking run with captured text output and a short timeout. Any
|
|
timeout, subprocess, or OS error is swallowed and reported as empty output,
|
|
so callers can split/scan the result unconditionally.
|
|
|
|
Args:
|
|
args: The command and its arguments.
|
|
timeout: Seconds before the command is killed.
|
|
|
|
Returns:
|
|
The command's standard output, or an empty string on any failure.
|
|
"""
|
|
try:
|
|
result = subprocess.run(
|
|
args,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout,
|
|
check=False,
|
|
)
|
|
except (subprocess.TimeoutExpired, subprocess.SubprocessError, OSError):
|
|
logger.debug("Command failed: %s", args, exc_info=True)
|
|
return ""
|
|
return result.stdout
|
|
|
|
|
|
def parse_cups_usb_uri(uri: str, info: dict[str, str]) -> None:
|
|
"""Extract the product and serial from a CUPS ``usb://`` URI into ``info``."""
|
|
parsed = urllib.parse.urlparse(uri)
|
|
info["product"] = urllib.parse.unquote(parsed.path.lstrip("/"))
|
|
query = urllib.parse.parse_qs(parsed.query)
|
|
if "serial" in query:
|
|
info["serial"] = query["serial"][0]
|
|
|
|
|
|
def printer_info_from_cups() -> dict[str, str]:
|
|
"""Return the Brother printer's model/serial as parsed from ``lpstat -v``."""
|
|
info: dict[str, str] = {"product": "", "serial": ""}
|
|
for line in run_command_text(["/usr/bin/lpstat", "-v"]).splitlines():
|
|
if "Brother" in line:
|
|
for part in line.split():
|
|
if part.startswith("usb://"):
|
|
parse_cups_usb_uri(part, info)
|
|
break
|
|
return info
|