mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-04 06:15:09 +00:00
chore(deprecation): retarget the 35 deprecations to 3.8.0 and add a usage scan (#681)
Moves the 35 @deprecated markers from 3.7.0 (already shipped with them in place) to 3.8.0, and adds scripts/plugin_api_usage.py plus the generated docs/DEPRECATIONS_3.8.md: who still calls or overrides each deprecated method across core, the monorepo and third-party plugins. Removes nothing. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Chuck
@@ -0,0 +1,778 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Who still calls or overrides the core methods marked ``@deprecated``?
|
||||
|
||||
A deprecated plugin-facing method may only be removed once nothing uses it,
|
||||
and plugins live in other repositories. This script answers the question for
|
||||
every method ``src/deprecation.py``'s decorator marks in core:
|
||||
|
||||
1. it lists the markers by parsing ``src/`` (so the list can never drift from
|
||||
the code);
|
||||
2. it scans, with the ``ast`` module, core itself (``src/``,
|
||||
``web_interface/``, ``scripts/``, the top-level ``*.py``; ``test/``
|
||||
separately), the official monorepo's ``plugins/`` directory, and every
|
||||
third-party plugin the monorepo's ``plugins.json`` lists with its own repo
|
||||
URL (shallow-cloned read-only into a cache directory);
|
||||
3. it reports, per method and per plugin, the calls and overrides it found,
|
||||
and a verdict: unused (safe to remove in the marker's release), still used
|
||||
(keep or migrate those plugins first), or needs review.
|
||||
|
||||
Matching is by method name, so it has to separate real uses from unrelated
|
||||
methods that happen to share the name (the weather plugin's own ``draw_sun``,
|
||||
say). Each hit is classified by what it is attached to:
|
||||
|
||||
* **call** -- ``<receiver>.name`` where the receiver is named like the owning
|
||||
object (``self.cache_manager``, ``display_manager``, ``plugin_manager`` ...,
|
||||
or a local alias assigned from one), or ``self``/``super()`` inside a class
|
||||
that subclasses the owner. Attribute references that are not called
|
||||
(``callback=cm.get_cache_metrics``) count too.
|
||||
* **override** -- ``def name`` in a class that subclasses the owner.
|
||||
* **review** -- ``<receiver>.name`` where the receiver says nothing about its
|
||||
type, or ``getattr(obj, "name")``. Possibly a real use; read the listed line.
|
||||
* **unrelated** -- ``self.name`` inside a class that defines ``name`` itself
|
||||
and does not subclass the owner, ``Klass.name`` where the same tree defines
|
||||
``Klass.name``, or ``def name`` in such a class: a name collision, not a use.
|
||||
* **internal** -- a hit inside the body of another deprecated core method
|
||||
(``draw_rain`` calling ``draw_cloud``): it keeps the method only as long as
|
||||
that caller is kept.
|
||||
|
||||
Only calls and overrides make a method "still used"; review hits make it
|
||||
"needs review"; hits in test files are listed but never block removal (a test
|
||||
that mocks a method does not need it to exist).
|
||||
|
||||
python3 scripts/plugin_api_usage.py # clone everything, print Markdown
|
||||
python3 scripts/plugin_api_usage.py --monorepo ../ledmatrix-plugins
|
||||
python3 scripts/plugin_api_usage.py --output docs/DEPRECATIONS_3.8.md
|
||||
python3 scripts/plugin_api_usage.py --format json
|
||||
|
||||
Nothing is ever written to the repositories it scans: the monorepo path is only
|
||||
read, and clones live in ``--cache-dir``.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import ast
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import subprocess # nosec B404 - list-form argv only, no shell # nosemgrep
|
||||
import sys
|
||||
import tempfile
|
||||
from collections import defaultdict
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Dict, Iterable, Iterator, List, Optional, Set, Tuple
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
MONOREPO_URL = "https://github.com/ChuckBuilds/ledmatrix-plugins"
|
||||
MONOREPO_SLUG = "chuckbuilds/ledmatrix-plugins"
|
||||
|
||||
#: Receiver names that mean "this is the owning core object". Compared against
|
||||
#: the last name in the receiver (``self.plugin_manager.cache_manager`` ->
|
||||
#: ``cache_manager``), lower-cased with leading underscores stripped.
|
||||
OWNER_RECEIVERS: Dict[str, Tuple[str, ...]] = {
|
||||
"CacheManager": ("cache_manager", "cache_mgr", "cachemanager", "cache", "cm"),
|
||||
"DisplayManager": ("display_manager", "display_mgr", "displaymanager", "display", "dm"),
|
||||
"FontManager": ("font_manager", "font_mgr", "fontmanager", "fonts", "fm"),
|
||||
"PluginManager": ("plugin_manager", "plugin_mgr", "pluginmanager", "pm"),
|
||||
}
|
||||
|
||||
#: Directories never scanned (vendored environments, VCS metadata, caches).
|
||||
SKIP_DIRS = {".git", "__pycache__", "node_modules", ".venv", "venv", "env",
|
||||
"site-packages", ".tox", ".mypy_cache", ".pytest_cache"}
|
||||
|
||||
CORE_DIRS = ("src", "web_interface", "scripts")
|
||||
CORE_TEST_DIRS = ("test",)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Markers
|
||||
|
||||
|
||||
@dataclass
|
||||
class Marker:
|
||||
owner: str # class name, e.g. "CacheManager"
|
||||
method: str
|
||||
removal: str
|
||||
alternative: Optional[str]
|
||||
module: str # e.g. "src.cache_manager"
|
||||
line: int
|
||||
|
||||
@property
|
||||
def key(self) -> str:
|
||||
return f"{self.owner}.{self.method}"
|
||||
|
||||
|
||||
def _decorator_name(node: ast.expr) -> Optional[str]:
|
||||
target = node.func if isinstance(node, ast.Call) else node
|
||||
if isinstance(target, ast.Name):
|
||||
return target.id
|
||||
if isinstance(target, ast.Attribute):
|
||||
return target.attr
|
||||
return None
|
||||
|
||||
|
||||
def find_markers(core_root: Path) -> List[Marker]:
|
||||
"""Every ``@deprecated(...)`` method under ``core_root/src``."""
|
||||
markers: List[Marker] = []
|
||||
for path in sorted((core_root / "src").rglob("*.py")):
|
||||
if path.name == "deprecation.py":
|
||||
continue
|
||||
tree = _parse(path)
|
||||
if tree is None:
|
||||
continue
|
||||
module = ".".join(path.relative_to(core_root).with_suffix("").parts)
|
||||
for cls in (n for n in ast.walk(tree) if isinstance(n, ast.ClassDef)):
|
||||
for fn in cls.body:
|
||||
if not isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
continue
|
||||
for dec in fn.decorator_list:
|
||||
if _decorator_name(dec) != "deprecated" or not isinstance(dec, ast.Call):
|
||||
continue
|
||||
args = [a.value if isinstance(a, ast.Constant) else None for a in dec.args]
|
||||
kw = {k.arg: k.value.value for k in dec.keywords
|
||||
if isinstance(k.value, ast.Constant)}
|
||||
removal = args[0] if args else kw.get("removal")
|
||||
alternative = args[1] if len(args) > 1 else kw.get("alternative")
|
||||
markers.append(Marker(cls.name, fn.name, str(removal), alternative,
|
||||
module, fn.lineno))
|
||||
return markers
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Scanning
|
||||
|
||||
|
||||
@dataclass
|
||||
class Hit:
|
||||
kind: str # call | override | review | unrelated | internal
|
||||
path: str
|
||||
line: int
|
||||
code: str
|
||||
test: bool
|
||||
via: Optional[str] = None # internal: the deprecated core method it sits in
|
||||
|
||||
|
||||
@dataclass
|
||||
class Source:
|
||||
"""One plugin (or core) tree to scan."""
|
||||
name: str
|
||||
group: str # core | core-tests | monorepo | third-party
|
||||
root: Optional[Path]
|
||||
error: Optional[str] = None
|
||||
hits: Dict[str, List[Hit]] = field(default_factory=lambda: defaultdict(list))
|
||||
files: int = 0 # Python files scanned
|
||||
|
||||
|
||||
def _parse(path: Path) -> Optional[ast.AST]:
|
||||
try:
|
||||
return ast.parse(path.read_text(encoding="utf-8", errors="replace"), str(path))
|
||||
except (SyntaxError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def _iter_py(root: Path) -> Iterator[Path]:
|
||||
for dirpath, dirnames, filenames in os.walk(root):
|
||||
dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS]
|
||||
for name in filenames:
|
||||
if name.endswith(".py"):
|
||||
yield Path(dirpath) / name
|
||||
|
||||
|
||||
def _is_test_path(rel: Path) -> bool:
|
||||
parts = [p.lower() for p in rel.parts]
|
||||
return (any(p in ("test", "tests") for p in parts[:-1])
|
||||
or parts[-1].startswith("test_") or parts[-1].endswith("_test.py")
|
||||
or parts[-1] == "conftest.py")
|
||||
|
||||
|
||||
def _terminal(node: ast.expr) -> Optional[str]:
|
||||
"""The last name in a receiver expression, or None if it has none."""
|
||||
if isinstance(node, ast.Name):
|
||||
return node.id
|
||||
if isinstance(node, ast.Attribute):
|
||||
return node.attr
|
||||
if isinstance(node, ast.Call):
|
||||
return _terminal(node.func)
|
||||
if isinstance(node, ast.Subscript):
|
||||
return _terminal(node.value)
|
||||
return None
|
||||
|
||||
|
||||
def _norm(name: Optional[str]) -> str:
|
||||
return (name or "").lstrip("_").lower()
|
||||
|
||||
|
||||
def _base_names(cls: ast.ClassDef) -> List[str]:
|
||||
return [t for t in (_terminal(b) for b in cls.bases) if t]
|
||||
|
||||
|
||||
class _Scanner(ast.NodeVisitor):
|
||||
"""Collect hits for every marked method name in one file."""
|
||||
|
||||
def __init__(self, markers: Dict[str, List[Marker]], lines: List[str],
|
||||
rel: str, test: bool, core_modules: Dict[str, str], module: Optional[str],
|
||||
local_definers: Dict[str, Set[str]], built: Dict[str, str]):
|
||||
self.markers = markers # method name -> markers with that name
|
||||
self.local_definers = local_definers # method name -> this tree's own classes/modules defining it
|
||||
self.built = built # ``x``/``self.x`` -> class it was built from in this file
|
||||
self.lines = lines
|
||||
self.rel = rel
|
||||
self.test = test
|
||||
self.core_modules = core_modules # owner class -> defining module (core only)
|
||||
self.module = module # this file's module when scanning core
|
||||
self.classes: List[ast.ClassDef] = []
|
||||
self.scope: List[ast.AST] = [] # enclosing classes and functions
|
||||
self.aliases: List[Dict[str, str]] = [{}] # local name -> owner class
|
||||
self.out: Dict[str, List[Hit]] = defaultdict(list)
|
||||
|
||||
# -- helpers
|
||||
def _code(self, node: ast.AST) -> str:
|
||||
line = self.lines[node.lineno - 1] if 0 < node.lineno <= len(self.lines) else ""
|
||||
return line.strip()[:160]
|
||||
|
||||
def _add(self, marker: Marker, kind: str, node: ast.AST) -> None:
|
||||
via = self._inside_deprecated()
|
||||
if via and kind != "unrelated":
|
||||
# Only reached through another deprecated method: goes when that does.
|
||||
kind = "internal"
|
||||
self.out[marker.key].append(Hit(kind, self.rel, node.lineno, self._code(node),
|
||||
self.test, via if kind == "internal" else None))
|
||||
|
||||
def _inside_deprecated(self) -> Optional[str]:
|
||||
"""``Owner.method`` when this node sits in a deprecated core method's body."""
|
||||
for i in range(len(self.scope) - 2, -1, -1):
|
||||
cls, fn = self.scope[i], self.scope[i + 1]
|
||||
if isinstance(cls, ast.ClassDef):
|
||||
if isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
for m in self.markers.get(fn.name, ()):
|
||||
if self._is_owner_class(cls, m.owner):
|
||||
return m.key
|
||||
return None
|
||||
return None
|
||||
|
||||
def _owner_for_receiver(self, name: Optional[str]) -> Optional[str]:
|
||||
n = _norm(name)
|
||||
for scope in reversed(self.aliases):
|
||||
if name in scope:
|
||||
return scope[name]
|
||||
for owner, receivers in OWNER_RECEIVERS.items():
|
||||
if n in receivers:
|
||||
return owner
|
||||
return None
|
||||
|
||||
def _is_owner_class(self, cls: ast.ClassDef, owner: str) -> bool:
|
||||
"""True for the real core class (only when scanning its own module)."""
|
||||
return (self.module is not None and cls.name == owner
|
||||
and self.core_modules.get(owner) == self.module)
|
||||
|
||||
def _subclasses(self, cls: ast.ClassDef, owner: str) -> bool:
|
||||
return owner in _base_names(cls)
|
||||
|
||||
def _class_defines(self, cls: ast.ClassDef, name: str) -> bool:
|
||||
return any(isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)) and n.name == name
|
||||
for n in cls.body)
|
||||
|
||||
# -- scopes
|
||||
def visit_ClassDef(self, node: ast.ClassDef) -> None:
|
||||
for fn in node.body:
|
||||
if isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)) and fn.name in self.markers:
|
||||
for m in self.markers[fn.name]:
|
||||
if self._is_owner_class(node, m.owner):
|
||||
continue # the definition itself
|
||||
kind = "override" if self._subclasses(node, m.owner) else "unrelated"
|
||||
self._add(m, kind, fn)
|
||||
self.classes.append(node)
|
||||
self.scope.append(node)
|
||||
self.generic_visit(node)
|
||||
self.scope.pop()
|
||||
self.classes.pop()
|
||||
|
||||
def _visit_function(self, node) -> None:
|
||||
self.aliases.append({})
|
||||
self.scope.append(node)
|
||||
self.generic_visit(node)
|
||||
self.scope.pop()
|
||||
self.aliases.pop()
|
||||
|
||||
visit_FunctionDef = _visit_function
|
||||
visit_AsyncFunctionDef = _visit_function
|
||||
|
||||
def visit_Assign(self, node: ast.Assign) -> None:
|
||||
# ``dm = self.display_manager`` makes ``dm.draw_sun()`` a call.
|
||||
owner = self._owner_for_receiver(_terminal(node.value))
|
||||
for target in node.targets:
|
||||
if isinstance(target, ast.Name) and owner:
|
||||
self.aliases[-1][target.id] = owner
|
||||
self.generic_visit(node)
|
||||
|
||||
# -- uses
|
||||
def visit_Attribute(self, node: ast.Attribute) -> None:
|
||||
if node.attr in self.markers:
|
||||
for m in self.markers[node.attr]:
|
||||
self._add(m, self._classify(node, m), node)
|
||||
self.generic_visit(node)
|
||||
|
||||
def _classify(self, node: ast.Attribute, m: Marker) -> str:
|
||||
recv = node.value
|
||||
cls = self.classes[-1] if self.classes else None
|
||||
is_self = isinstance(recv, ast.Name) and recv.id in ("self", "cls")
|
||||
is_super = (isinstance(recv, ast.Call) and isinstance(recv.func, ast.Name)
|
||||
and recv.func.id == "super")
|
||||
if is_self or is_super:
|
||||
if cls is not None and (self._is_owner_class(cls, m.owner) or self._subclasses(cls, m.owner)):
|
||||
return "call"
|
||||
if cls is not None and self._class_defines(cls, m.method):
|
||||
return "unrelated"
|
||||
return "review"
|
||||
name = _terminal(recv)
|
||||
if self._owner_for_receiver(name) == m.owner:
|
||||
return "call"
|
||||
definers = self.local_definers.get(m.method, ())
|
||||
if name in definers or self.built.get(name or "") in definers:
|
||||
# e.g. the weather plugin's WeatherIcons.draw_sun, or
|
||||
# self._strategy_component = CacheStrategy(); ...get_sport_live_interval()
|
||||
return "unrelated"
|
||||
return "review"
|
||||
|
||||
def visit_Call(self, node: ast.Call) -> None:
|
||||
func = node.func
|
||||
if (isinstance(func, ast.Name) and func.id in ("getattr", "hasattr", "setattr", "delattr")
|
||||
and len(node.args) >= 2 and isinstance(node.args[1], ast.Constant)
|
||||
and node.args[1].value in self.markers):
|
||||
for m in self.markers[node.args[1].value]:
|
||||
owner = self._owner_for_receiver(_terminal(node.args[0]))
|
||||
self._add(m, "call" if owner == m.owner else "review", node)
|
||||
self.generic_visit(node)
|
||||
|
||||
|
||||
def _built_from(tree: ast.AST) -> Dict[str, str]:
|
||||
"""``{name: Class}`` for every ``name = Class(...)`` / ``self.name = Class(...)``."""
|
||||
built: Dict[str, str] = {}
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Assign) and isinstance(node.value, ast.Call):
|
||||
cls = _terminal(node.value.func)
|
||||
for target in node.targets:
|
||||
name = _terminal(target) if isinstance(target, (ast.Name, ast.Attribute)) else None
|
||||
if name and cls:
|
||||
built[name] = cls
|
||||
return built
|
||||
|
||||
|
||||
def scan_tree(source: Source, roots: Iterable[Path], base: Path, markers: List[Marker],
|
||||
core: bool, test_override: Optional[bool] = None,
|
||||
definer_roots: Iterable[Path] = ()) -> None:
|
||||
by_name: Dict[str, List[Marker]] = defaultdict(list)
|
||||
for m in markers:
|
||||
by_name[m.method].append(m)
|
||||
core_modules = {m.owner: m.module for m in markers}
|
||||
owners = {m.owner for m in markers}
|
||||
files: List[Tuple[Path, str, Optional[ast.AST]]] = []
|
||||
for root in roots:
|
||||
if not root.exists():
|
||||
continue
|
||||
for path in ([root] if root.is_file() else sorted(_iter_py(root))):
|
||||
source.files += 1
|
||||
text = path.read_text(encoding="utf-8", errors="replace")
|
||||
if any(name in text for name in by_name):
|
||||
files.append((path, text, _parse(path)))
|
||||
|
||||
# Classes (and modules) in this tree with their own method of a marked
|
||||
# name, so ``WeatherIcons.draw_sun()`` is recognised as theirs.
|
||||
local_definers: Dict[str, Set[str]] = defaultdict(set)
|
||||
definer_files = list(files)
|
||||
for root in definer_roots:
|
||||
for path in (sorted(_iter_py(root)) if root.is_dir() else ()):
|
||||
text = path.read_text(encoding="utf-8", errors="replace")
|
||||
if any(name in text for name in by_name):
|
||||
definer_files.append((path, text, _parse(path)))
|
||||
for path, _, tree in definer_files:
|
||||
for node in (tree.body if tree is not None else ()):
|
||||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name in by_name:
|
||||
local_definers[node.name].add(path.stem)
|
||||
for cls in (n for n in ast.walk(tree) if isinstance(n, ast.ClassDef)) if tree else ():
|
||||
if cls.name in owners and core:
|
||||
continue
|
||||
if owners & set(_base_names(cls)):
|
||||
continue
|
||||
for fn in cls.body:
|
||||
if isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)) and fn.name in by_name:
|
||||
local_definers[fn.name].add(cls.name)
|
||||
|
||||
for path, text, tree in files:
|
||||
rel = path.relative_to(base)
|
||||
test = _is_test_path(rel) if test_override is None else test_override
|
||||
if tree is None:
|
||||
# Unparseable (Python 2, a template ...): fall back to text, as review.
|
||||
for no, line in enumerate(text.splitlines(), 1):
|
||||
for name in by_name:
|
||||
if re.search(rf"{re.escape(name)}", line):
|
||||
for m in by_name[name]:
|
||||
source.hits[m.key].append(
|
||||
Hit("review", rel.as_posix(), no, line.strip()[:160], test))
|
||||
continue
|
||||
module = ".".join(rel.with_suffix("").parts) if core else None
|
||||
scanner = _Scanner(by_name, text.splitlines(), rel.as_posix(), test,
|
||||
core_modules, module, local_definers, _built_from(tree))
|
||||
scanner.visit(tree)
|
||||
for key, hits in scanner.out.items():
|
||||
source.hits[key].extend(hits)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Fetching plugin trees (read-only)
|
||||
|
||||
|
||||
def _git(*args: str, cwd: Optional[Path] = None) -> subprocess.CompletedProcess:
|
||||
# Never stop to ask for credentials: a deleted or private plugin repo
|
||||
# should be reported as not scanned, not hang the scan.
|
||||
env = {**os.environ, "GIT_TERMINAL_PROMPT": "0"}
|
||||
return subprocess.run( # nosec B603 B607 - list-form git argv, no shell; URLs follow "--" # nosemgrep
|
||||
["git", *args], cwd=cwd, capture_output=True, text=True,
|
||||
encoding="utf-8", errors="replace", timeout=300, env=env)
|
||||
|
||||
|
||||
def _rmtree(path: Path) -> None:
|
||||
"""Delete a clone; git marks pack files read-only, which Windows refuses to delete."""
|
||||
import shutil
|
||||
import stat
|
||||
|
||||
def retry(func, target, _exc):
|
||||
os.chmod(target, stat.S_IWRITE)
|
||||
func(target)
|
||||
|
||||
if sys.version_info >= (3, 12):
|
||||
shutil.rmtree(path, onexc=retry)
|
||||
else:
|
||||
shutil.rmtree(path, onerror=retry)
|
||||
|
||||
|
||||
def shallow_clone(url: str, branch: Optional[str], dest: Path, reuse: bool) -> Optional[str]:
|
||||
"""Clone ``url`` into ``dest`` (depth 1), replacing any earlier clone.
|
||||
|
||||
Returns an error string, or None on success.
|
||||
"""
|
||||
if reuse and (dest / ".git").exists():
|
||||
return None
|
||||
if dest.exists():
|
||||
_rmtree(dest)
|
||||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||||
args = ["-c", "core.longpaths=true", "clone", "--quiet", "--depth", "1"]
|
||||
if branch:
|
||||
args += ["--branch", branch]
|
||||
# "--" ends option parsing: a registry URL starting with "-" (for example
|
||||
# "--upload-pack=...") is then only ever a repository argument.
|
||||
result = _git(*args, "--", url, str(dest))
|
||||
if result.returncode != 0 and branch:
|
||||
result = _git("-c", "core.longpaths=true", "clone", "--quiet", "--depth", "1",
|
||||
"--", url, str(dest))
|
||||
if result.returncode != 0:
|
||||
lines = (result.stderr or result.stdout).strip().splitlines()
|
||||
return lines[-1] if lines else "git clone failed"
|
||||
return None
|
||||
|
||||
|
||||
def _head(path: Path, branch: bool = True) -> str:
|
||||
"""Commit (and branch) of a checkout, via read-only git calls."""
|
||||
rev = _git("--no-optional-locks", "rev-parse", "--short=8", "HEAD", cwd=path)
|
||||
if rev.returncode != 0:
|
||||
return "unknown revision"
|
||||
if not branch:
|
||||
return rev.stdout.strip()
|
||||
ref = _git("--no-optional-locks", "rev-parse", "--abbrev-ref", "HEAD", cwd=path)
|
||||
return f"{ref.stdout.strip()} @ {rev.stdout.strip()}"
|
||||
|
||||
|
||||
def _plugin_id(plugin_dir: Path) -> str:
|
||||
try:
|
||||
return json.loads((plugin_dir / "manifest.json").read_text(encoding="utf-8"))["id"]
|
||||
except (OSError, ValueError, KeyError, TypeError):
|
||||
return plugin_dir.name
|
||||
|
||||
|
||||
def _is_monorepo(url: str) -> bool:
|
||||
return MONOREPO_SLUG in url.lower().rstrip("/").removesuffix(".git")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Report
|
||||
|
||||
|
||||
def verdicts(markers: List[Marker], sources: List[Source]) -> Dict[str, Tuple[str, str]]:
|
||||
"""``{Owner.method: (status, text)}`` across every source.
|
||||
|
||||
An *internal* hit (a call from inside another deprecated method) keeps a
|
||||
method only while that caller is itself kept, so statuses are resolved
|
||||
until they stop changing.
|
||||
"""
|
||||
failed = [s.name for s in sources if s.error]
|
||||
status: Dict[str, Tuple[str, str]] = {}
|
||||
for _ in range(len(markers) + 1):
|
||||
changed = False
|
||||
for m in markers:
|
||||
used, review = [], []
|
||||
for s in sources:
|
||||
live = [h for h in s.hits.get(m.key, []) if not h.test]
|
||||
if any(h.kind in ("call", "override") for h in live) or any(
|
||||
h.kind == "internal" and status.get(h.via, ("",))[0] == "used"
|
||||
for h in live):
|
||||
used.append(s.name)
|
||||
elif any(h.kind == "review" for h in live) or any(
|
||||
h.kind == "internal" and status.get(h.via, ("",))[0] == "review"
|
||||
for h in live):
|
||||
review.append(s.name)
|
||||
if used:
|
||||
new = ("used", f"still used by {', '.join(used)} — keep or migrate first")
|
||||
elif review:
|
||||
new = ("review", f"needs review: possible use in {', '.join(review)}")
|
||||
elif failed:
|
||||
new = ("unknown", f"not proven unused: {len(failed)} plugin(s) could not be scanned")
|
||||
else:
|
||||
new = ("unused", f"unused — safe to remove in {m.removal}")
|
||||
if status.get(m.key) != new:
|
||||
status[m.key] = new
|
||||
changed = True
|
||||
if not changed:
|
||||
break
|
||||
return status
|
||||
|
||||
|
||||
def _counts(hits: List[Hit]) -> Dict[str, int]:
|
||||
c: Dict[str, int] = defaultdict(int)
|
||||
for h in hits:
|
||||
c[("test " if h.test else "") + h.kind] += 1
|
||||
return c
|
||||
|
||||
|
||||
def _usage_cell(marker: Marker, sources: List[Source], kinds: Tuple[str, ...]) -> str:
|
||||
parts = []
|
||||
for s in sources:
|
||||
c = _counts(s.hits.get(marker.key, []))
|
||||
bits = [f"{c[k]} {k}{'s' if c[k] != 1 else ''}" for k in kinds if c[k]]
|
||||
if bits:
|
||||
parts.append(f"{s.name} ({', '.join(bits)})")
|
||||
return "; ".join(parts) or "—"
|
||||
|
||||
|
||||
def render_markdown(markers: List[Marker], sources: List[Source], meta: Dict[str, str]) -> str:
|
||||
out: List[str] = []
|
||||
w = out.append
|
||||
w("# Deprecated plugin APIs: usage scan")
|
||||
w("")
|
||||
w("Generated by `scripts/plugin_api_usage.py` — do not edit by hand; re-run it "
|
||||
"(see [How to re-run](#how-to-re-run)).")
|
||||
w("")
|
||||
w(f"- Scanned: {meta['date']}, core {meta['core_version']}")
|
||||
w(f"- Monorepo: {meta['monorepo']}")
|
||||
w(f"- Third-party plugins: {meta['third_party']}")
|
||||
failed = [s for s in sources if s.error]
|
||||
if failed:
|
||||
w("- **Not scanned:** " + "; ".join(f"{s.name} ({s.error})" for s in failed))
|
||||
w("")
|
||||
status = verdicts(markers, sources)
|
||||
tally: Dict[str, int] = defaultdict(int)
|
||||
for st, _ in status.values():
|
||||
tally[st] += 1
|
||||
w(f"**{len(markers)} deprecated methods: {tally['unused']} unused, "
|
||||
f"{tally['used']} still used, {tally['review']} need review"
|
||||
+ (f", {tally['unknown']} not proven" if tally["unknown"] else "") + ".**")
|
||||
w("")
|
||||
w("Counted per plugin: a *call* is `<receiver>.method` on an object named like "
|
||||
"the owner (`cache_manager`, `display_manager`, `font_manager`, `plugin_manager`), "
|
||||
"or on `self`/`super()` in a subclass; an *override* is `def method` in a subclass "
|
||||
"of the owner. *Review* hits are `.method` on a receiver whose type the scan cannot "
|
||||
"tell. *Internal* hits sit inside another deprecated core method and go with it. "
|
||||
"*Unrelated* hits are a different class's own method with the same name "
|
||||
"(a name collision), and never block removal; neither do hits in test files.")
|
||||
w("")
|
||||
w("| Method | Removal | Core | Plugins (calls / overrides) | Name collisions & tests | Verdict |")
|
||||
w("|---|---|---|---|---|---|")
|
||||
core_sources = [s for s in sources if s.group in ("core", "core-tests")]
|
||||
plugin_sources = [s for s in sources if s.group not in ("core", "core-tests")]
|
||||
for m in markers:
|
||||
core = _usage_cell(m, core_sources, ("call", "override", "review", "internal",
|
||||
"test call", "test override", "test review",
|
||||
"test internal"))
|
||||
plugins = _usage_cell(m, plugin_sources, ("call", "override", "review"))
|
||||
other = _usage_cell(m, plugin_sources, ("unrelated", "test call", "test override",
|
||||
"test review", "test unrelated"))
|
||||
w(f"| `{m.key}` | {m.removal} | {core} | {plugins} | {other} | {status[m.key][1]} |")
|
||||
w("")
|
||||
|
||||
groups = [("unused", "Unused — safe to remove"), ("used", "Still used — keep or migrate first"),
|
||||
("review", "Needs review"), ("unknown", "Not proven unused")]
|
||||
for st, title in groups:
|
||||
names = [m.key for m in markers if status[m.key][0] == st]
|
||||
if names:
|
||||
w(f"## {title} ({len(names)})")
|
||||
w("")
|
||||
w(", ".join(f"`{n}`" for n in names))
|
||||
w("")
|
||||
|
||||
detail = [(m, s, h) for m in markers for s in sources
|
||||
for h in s.hits.get(m.key, []) if h.kind != "unrelated" or not h.test]
|
||||
if detail:
|
||||
w("## Every hit")
|
||||
w("")
|
||||
w("File paths are relative to the plugin's directory (core: the repo root).")
|
||||
w("")
|
||||
w("| Method | Where | File:line | Kind | Code |")
|
||||
w("|---|---|---|---|---|")
|
||||
for m, s, h in detail:
|
||||
kind = ("test " if h.test else "") + h.kind
|
||||
if h.via:
|
||||
kind += f" (in `{h.via}`)"
|
||||
code = h.code.replace("|", "\\|").replace("`", "'")
|
||||
w(f"| `{m.key}` | {s.name} | {h.path}:{h.line} | {kind} | `{code}` |")
|
||||
w("")
|
||||
|
||||
w("## Sources scanned")
|
||||
w("")
|
||||
w("| Source | Group | Python files | Hits |")
|
||||
w("|---|---|---|---|")
|
||||
for s in sources:
|
||||
n = sum(len(v) for v in s.hits.values())
|
||||
files = f"not scanned: {s.error}" if s.error else str(s.files)
|
||||
w(f"| {s.name} | {s.group} | {files} | {n} |")
|
||||
w("")
|
||||
|
||||
w("## How to re-run")
|
||||
w("")
|
||||
w("```bash")
|
||||
w("# Clones the monorepo and each third-party plugin (depth 1) into a temp cache:")
|
||||
w("python3 scripts/plugin_api_usage.py --output docs/DEPRECATIONS_3.8.md")
|
||||
w("# Or scan a local monorepo checkout (read only) instead of cloning it:")
|
||||
w("python3 scripts/plugin_api_usage.py --monorepo ../ledmatrix-plugins")
|
||||
w("```")
|
||||
w("")
|
||||
w("Before removing a method in its release, re-run the scan against the current "
|
||||
"monorepo and registry: a plugin added since this file was generated may have "
|
||||
"started calling it. Remove only methods the fresh scan reports unused; move "
|
||||
"the rest to a later release (the test in `test/test_deprecation.py` fails "
|
||||
"while a marker names a release at or below `src.__version__`).")
|
||||
w("")
|
||||
return "\n".join(out)
|
||||
|
||||
|
||||
def render_json(markers: List[Marker], sources: List[Source], meta: Dict[str, str]) -> str:
|
||||
data = {"meta": meta, "sources": [{"name": s.name, "group": s.group, "error": s.error}
|
||||
for s in sources], "methods": []}
|
||||
status = verdicts(markers, sources)
|
||||
for m in markers:
|
||||
st, text = status[m.key]
|
||||
data["methods"].append({
|
||||
"method": m.key, "module": m.module, "removal": m.removal,
|
||||
"alternative": m.alternative, "status": st, "verdict": text,
|
||||
"hits": [{"source": s.name, **h.__dict__} for s in sources
|
||||
for h in s.hits.get(m.key, [])],
|
||||
})
|
||||
return json.dumps(data, indent=2)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def main(argv: Optional[List[str]] = None) -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
|
||||
parser.add_argument("--monorepo", type=Path,
|
||||
help="local ledmatrix-plugins checkout to scan (read only); "
|
||||
"default: shallow-clone its main branch")
|
||||
parser.add_argument("--registry", type=Path,
|
||||
help="plugins.json to read third-party plugins from "
|
||||
"(default: the monorepo's)")
|
||||
parser.add_argument("--cache-dir", type=Path,
|
||||
default=Path(tempfile.gettempdir()) / "ledmatrix-plugin-api-usage",
|
||||
help="where clones go (default: %(default)s)")
|
||||
parser.add_argument("--reuse-cache", action="store_true",
|
||||
help="scan clones already in --cache-dir instead of re-cloning "
|
||||
"(offline re-runs; the report may then be stale)")
|
||||
parser.add_argument("--no-third-party", action="store_true",
|
||||
help="skip third-party plugins (the report then cannot prove anything unused)")
|
||||
parser.add_argument("--format", choices=("md", "json"), default="md")
|
||||
parser.add_argument("--output", type=Path, help="write the report here instead of stdout")
|
||||
args = parser.parse_args(argv)
|
||||
if hasattr(sys.stdout, "reconfigure"):
|
||||
sys.stdout.reconfigure(encoding="utf-8")
|
||||
|
||||
markers = find_markers(REPO_ROOT)
|
||||
if not markers:
|
||||
print("No @deprecated markers found in src/.", file=sys.stderr)
|
||||
return 0
|
||||
|
||||
sys.path.insert(0, str(REPO_ROOT))
|
||||
try:
|
||||
from src import __version__ as core_version
|
||||
except Exception: # noqa: BLE001 -- reporting only
|
||||
core_version = "unknown"
|
||||
|
||||
sources: List[Source] = []
|
||||
core = Source("core", "core", REPO_ROOT)
|
||||
scan_tree(core, [REPO_ROOT / d for d in CORE_DIRS] + sorted(REPO_ROOT.glob("*.py")),
|
||||
REPO_ROOT, markers, core=True)
|
||||
core_tests = Source("core tests", "core-tests", REPO_ROOT)
|
||||
scan_tree(core_tests, [REPO_ROOT / d for d in CORE_TEST_DIRS], REPO_ROOT, markers,
|
||||
core=True, test_override=True,
|
||||
definer_roots=[REPO_ROOT / d for d in CORE_DIRS])
|
||||
sources += [core, core_tests]
|
||||
|
||||
# Monorepo
|
||||
if args.monorepo:
|
||||
mono = args.monorepo.resolve()
|
||||
mono_desc = f"local checkout `{mono.name}` ({_head(mono)})"
|
||||
else:
|
||||
mono = args.cache_dir / "ledmatrix-plugins"
|
||||
err = shallow_clone(MONOREPO_URL, "main", mono, args.reuse_cache)
|
||||
if err:
|
||||
print(f"Could not clone the monorepo: {err}", file=sys.stderr)
|
||||
return 1
|
||||
mono_desc = f"[ChuckBuilds/ledmatrix-plugins]({MONOREPO_URL}) ({_head(mono)})"
|
||||
plugins_dir = mono / "plugins"
|
||||
mono_dirs = sorted(p for p in plugins_dir.iterdir() if p.is_dir()) if plugins_dir.is_dir() else []
|
||||
for d in mono_dirs:
|
||||
s = Source(_plugin_id(d), "monorepo", d)
|
||||
scan_tree(s, [d], d, markers, core=False)
|
||||
sources.append(s)
|
||||
mono_desc += f", {len(mono_dirs)} plugins"
|
||||
|
||||
# Third-party plugins from the registry
|
||||
registry = args.registry or (mono / "plugins.json")
|
||||
third: List[dict] = []
|
||||
try:
|
||||
reg = json.loads(registry.read_text(encoding="utf-8"))
|
||||
entries = reg["plugins"] if isinstance(reg, dict) else reg
|
||||
third = [e for e in entries if e.get("repo") and not _is_monorepo(e["repo"])]
|
||||
except (OSError, ValueError, KeyError) as exc:
|
||||
print(f"Could not read {registry}: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
if args.no_third_party:
|
||||
tp_desc = "skipped (--no-third-party)"
|
||||
else:
|
||||
for e in third:
|
||||
dest = args.cache_dir / "third-party" / re.sub(r"[^\w.-]", "_", e["id"])
|
||||
err = shallow_clone(e["repo"], e.get("branch") or None, dest, args.reuse_cache)
|
||||
root = dest / e["plugin_path"] if e.get("plugin_path") else dest
|
||||
s = Source(e["id"], "third-party", root, error=err)
|
||||
if not err:
|
||||
scan_tree(s, [root], root, markers, core=False)
|
||||
sources.append(s)
|
||||
tp_desc = (f"{len(third)} with their own repo in `plugins.json` "
|
||||
f"({', '.join(e['id'] for e in third)})")
|
||||
|
||||
meta = {
|
||||
"date": datetime.now(timezone.utc).strftime("%Y-%m-%d"),
|
||||
"core_version": core_version,
|
||||
"core_rev": _head(REPO_ROOT, branch=False),
|
||||
"monorepo": mono_desc,
|
||||
"third_party": tp_desc,
|
||||
}
|
||||
report = (render_json if args.format == "json" else render_markdown)(markers, sources, meta)
|
||||
if args.output:
|
||||
args.output.write_text(report, encoding="utf-8", newline="\n")
|
||||
print(f"Wrote {args.output}", file=sys.stderr)
|
||||
else:
|
||||
sys.stdout.write(report)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user