From b8c01c69fbc646796c7b160aba637e496bfd9b80 Mon Sep 17 00:00:00 2001 From: Chuck <33324927+ChuckBuilds@users.noreply.github.com> Date: Mon, 28 Sep 2026 15:01:35 -0400 Subject: [PATCH] ci: mypy ratchet -- keep type-clean modules clean (71 modules, 536 -> 442 errors) (#661) * ci: mypy ratchet -- keep type-clean modules clean mypy-clean.txt lists the 71 modules under src/ that type-check clean; scripts/check_types.py runs mypy (--follow-imports=silent) on exactly those files and fails on any error or a missing/unsorted/duplicate entry. A new "Type check (mypy ratchet)" CI job runs it with mypy 1.20.2 and pinned stubs; the manual pre-commit mypy hook now runs the same script (a local hook, so mypy sees the installed requirements like CI does). 35 modules were made clean with annotation-only fixes: hints, typing.cast, TYPE_CHECKING imports, implicit-Optional defaults made explicit, and annotations widened (never guards removed) where mypy called a defensive isinstance check unreachable. No runtime behaviour change. mypy.ini: numpy and orjson are treated as Any (follow_imports=skip, also for stubs). numpy 2.3+ stubs use 3.12 `type` statements that mypy won't parse at python_version 3.10, and orjson is optional, so seeing its stubs made the result depend on whether it was installed. Co-Authored-By: Claude Opus 5.5 * chore: annotate check_types.py's list-form mypy subprocess Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- .github/workflows/test.yml | 28 +++++++ .pre-commit-config.yaml | 20 ++--- CHANGELOG.md | 1 + CONTRIBUTING.md | 13 ++- mypy-clean.txt | 81 +++++++++++++++++++ mypy.ini | 17 ++++ scripts/check_types.py | 99 +++++++++++++++++++++++ src/backup_manager.py | 3 +- src/base_odds_manager.py | 8 +- src/cache/memory_cache.py | 9 ++- src/common/api_helper.py | 18 +++-- src/common/espn_dates.py | 8 +- src/common/frame_timing.py | 10 ++- src/common/logo_helper.py | 4 +- src/common/render_gate.py | 6 +- src/common/scroll_config.py | 9 ++- src/common/sports_card.py | 3 +- src/config_service.py | 3 +- src/deprecation.py | 2 +- src/device_location.py | 7 +- src/display_geometry.py | 1 + src/display_manager.py | 7 +- src/dynamic_team_resolver.py | 6 +- src/exceptions.py | 12 +-- src/logging_config.py | 5 +- src/logo_downloader.py | 7 +- src/plugin_system/compatibility.py | 8 +- src/plugin_system/plugin_dirs.py | 11 +-- src/plugin_system/plugin_executor.py | 4 +- src/plugin_system/plugin_loader.py | 1 + src/plugin_system/resource_monitor.py | 8 +- src/plugin_system/saved_repositories.py | 4 +- src/plugin_system/state_reconciliation.py | 9 ++- src/plugin_system/testing/loading.py | 8 +- src/plugin_system/testing/mocks.py | 10 +-- src/vegas_mode/coordinator.py | 6 +- src/vegas_mode/stream_manager.py | 4 +- src/web_interface/api_helpers.py | 2 +- src/web_interface/config_arrays.py | 4 +- src/web_interface/error_handler.py | 2 +- src/web_interface/errors.py | 2 +- src/web_interface/secret_helpers.py | 6 +- 42 files changed, 372 insertions(+), 104 deletions(-) create mode 100644 mypy-clean.txt create mode 100644 scripts/check_types.py diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index ffabca15..39179b3f 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -112,3 +112,31 @@ jobs: BASE: http://127.0.0.1:5000 REQUIRE_DOM: "1" run: node test/js/run_all.js + + type-check: + name: Type check (mypy ratchet) + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2 + with: + persist-credentials: false + + - uses: actions/setup-python@0b93645e9fea7318ecaed2b359559ac225c90a2b # v5.3.0 + with: + python-version: "3.12" + cache: pip + + # The runtime requirements are installed so mypy sees the real types of + # PIL, requests, psutil and friends -- missing, they'd be Any and the + # result would differ from a developer's machine. mypy and the stubs are + # pinned so a new release can't turn this red without a code change. + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -r requirements.txt -r web_interface/requirements.txt + pip install mypy==1.20.2 types-requests==2.33.0.20260906 types-pytz==2026.4.0.20260926 + + # mypy on exactly the modules in mypy-clean.txt; fails on any error in + # them, or if a listed file is missing. See CONTRIBUTING.md. + - name: Run mypy on the ratchet list + run: python scripts/check_types.py diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 1ff11b58..8230d4bb 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -37,19 +37,21 @@ repos: types: [python] pass_filenames: false - # Manual only: src/ still has ~500 type errors, so running this on every - # commit would block every contributor. Run it with + # The mypy ratchet -- the same check as CI's "Type check (mypy ratchet)" + # job: mypy on exactly the modules listed in mypy-clean.txt. Run it with # pre-commit run mypy --hook-stage manual - # and don't add new errors in the files you touch. mypy.ini's `files = src` - # is what gives it a target, since pass_filenames is off. - - repo: https://github.com/pre-commit/mirrors-mypy - rev: v1.8.0 + # A local hook rather than mirrors-mypy so mypy sees the packages installed + # from requirements.txt, as CI does; an isolated hook env without them types + # PIL, requests and friends as Any and reports different errors. Needs + # mypy==1.20.2 (the version CI pins) in the environment you commit from. + - repo: local hooks: - id: mypy - additional_dependencies: [types-requests, types-pytz] - args: [--ignore-missing-imports, --no-error-summary] + name: mypy (ratchet, mypy-clean.txt) + entry: python scripts/check_types.py + language: system pass_filenames: false - files: ^src/ + always_run: true stages: [manual] - repo: https://github.com/PyCQA/bandit diff --git a/CHANGELOG.md b/CHANGELOG.md index 25b9f388..cd2ce1bc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,7 @@ accepts both, but the store flags the old spelling as deprecated ## Unreleased +- A mypy ratchet in CI. `mypy-clean.txt` lists the 71 modules under `src/` that type-check clean, and the new "Type check (mypy ratchet)" job runs `python scripts/check_types.py` (mypy 1.20.2 on exactly those files) so they stay clean; add a module when you make it clean (see CONTRIBUTING.md). The manual pre-commit `mypy` hook runs the same script. 35 modules were made clean for it with annotation-only fixes, no behaviour change. Their public signatures only widened (`declared_min_version()` now says it returns the manifest's value as-is, `Any`); `DynamicTeamResolver._rankings_cache` is annotated as the abbreviation-to-rank dict it holds. `mypy.ini` treats numpy and orjson as `Any`, so it parses with `python_version = 3.10` against numpy 2.3+ stubs and gives the same result whether orjson is installed or not. - CI runs the web UI's DOM test suites (jsdom against the real server-rendered pages and API) in a new **Web UI JS tests** job, with the web interface started in emulator mode; `REQUIRE_DOM=1` makes a suite that can't run fail instead of being skipped. Two suites that had gone stale were fixed: the Tools suite now installs `LEDEscape` the way `base.html` does and supplies sample Starlark apps when the server has none, and the Store suite no longer assumes the registry has 48 plugins or fewer. - `src/plugin_system/store_manager.py` (2,977 lines) is split into mixins: `store_registry.py` (registry, GitHub metadata, search, manifest validation), `store_install.py` (install paths and dependencies) and `store_update.py` (updates, rollback, local git state). `PluginStoreManager` is still imported from `store_manager.py` and has exactly the same methods and attributes; every method body is byte-identical. - `BaseOddsManager.get_odds()` no longer returns the cached "no odds" marker (`{"no_odds": True}`) as if it were odds. A game ESPN had no odds for is cached that way so it isn't re-requested every update; on the next update the cache hit handed the marker back, and callers saw a truthy dict. It now returns `None` for it, on the cache hit and in the stale-cache fallback after a failed fetch, as the plugins' bundled copies already did. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index dd0522ca..7f3a7eb2 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -61,10 +61,15 @@ integration tests. `flake8` (E9, F63, F7, F82 plus bugbear `B` checks), `bandit`, and `gitleaks` — install the CLI with `python -m pip install pre-commit`, then run - `pre-commit install` so they run on every commit. `mypy` on - `src/` is a manual hook while existing type errors are paid down - (`pre-commit run mypy --hook-stage manual`): please don't add new - errors in the files you touch. HTML/JS in + `pre-commit install` so they run on every commit. Type checking + is a ratchet while the existing mypy errors in `src/` are paid + down: `mypy-clean.txt` lists the modules that type-check clean, and + CI runs `python scripts/check_types.py` (also the manual hook + `pre-commit run mypy --hook-stage manual`) to keep every listed + module clean. When you make another module clean, add it to the + list (sorted); don't take one off to get CI green. Keep type fixes + annotation-only where you can -- widen a hint rather than delete a + defensive runtime check mypy calls unreachable. HTML/JS in `web_interface/` follows the patterns already in `templates/v3/` and `static/v3/`. 5. **Update documentation** alongside code changes. If you add a diff --git a/mypy-clean.txt b/mypy-clean.txt new file mode 100644 index 00000000..a78faf48 --- /dev/null +++ b/mypy-clean.txt @@ -0,0 +1,81 @@ +# The mypy ratchet: modules that type-check clean, one path per line, sorted. +# +# CI ("Type check (mypy ratchet)") runs `python scripts/check_types.py`, which +# runs mypy on exactly these files (imports followed silently, so errors in an +# unlisted module they import don't count) and fails on any error, so a listed +# module stays clean. Most of src/ isn't clean yet. When you make a module +# clean, add it here. Don't take a module off to get CI green -- fix the error +# (annotation-only where you can: hints, typing.cast, TYPE_CHECKING imports; +# widen an annotation rather than delete a defensive runtime check). + +src/__init__.py +src/adaptive_images.py +src/auto_update_setup.py +src/backup_manager.py +src/base_odds_manager.py +src/cache/__init__.py +src/cache/cache_metrics.py +src/cache/cache_strategy.py +src/cache/memory_cache.py +src/common/__init__.py +src/common/api_helper.py +src/common/bdf_font.py +src/common/espn_dates.py +src/common/font_layout.py +src/common/frame_timing.py +src/common/json_body.py +src/common/logo_helper.py +src/common/path_safety.py +src/common/permission_utils.py +src/common/render_gate.py +src/common/scroll_config.py +src/common/snapshot_policy.py +src/common/sports_card.py +src/common/sports_scroll.py +src/config_service.py +src/core_config_keys.py +src/deprecation.py +src/device_location.py +src/display_geometry.py +src/dynamic_team_resolver.py +src/exceptions.py +src/font_usage.py +src/logging_config.py +src/logo_downloader.py +src/matrix_support.py +src/pi5_matrix_support.py +src/plugin_system/__init__.py +src/plugin_system/compatibility.py +src/plugin_system/operation_history.py +src/plugin_system/operation_queue.py +src/plugin_system/operation_types.py +src/plugin_system/plugin_dirs.py +src/plugin_system/plugin_executor.py +src/plugin_system/plugin_health.py +src/plugin_system/plugin_loader.py +src/plugin_system/plugin_state.py +src/plugin_system/repo_urls.py +src/plugin_system/resource_monitor.py +src/plugin_system/saved_repositories.py +src/plugin_system/schema_manager.py +src/plugin_system/state_reconciliation.py +src/plugin_system/testing/__init__.py +src/plugin_system/testing/bounds_display_manager.py +src/plugin_system/testing/loading.py +src/plugin_system/testing/mocks.py +src/plugin_system/testing/plugin_test_base.py +src/plugin_system/testing/sizes.py +src/redaction.py +src/scan_order.py +src/startup_validator.py +src/vegas_mode/__init__.py +src/vegas_mode/config.py +src/vegas_mode/coordinator.py +src/vegas_mode/geometry.py +src/vegas_mode/stream_manager.py +src/web_interface/api_helpers.py +src/web_interface/config_arrays.py +src/web_interface/error_handler.py +src/web_interface/errors.py +src/web_interface/secret_helpers.py +src/web_interface/validators.py diff --git a/mypy.ini b/mypy.ini index b77e3e7b..603521c6 100644 --- a/mypy.ini +++ b/mypy.ini @@ -101,3 +101,20 @@ ignore_missing_imports = True [mypy-spotipy.*] ignore_missing_imports = True + +# numpy's own stubs (numpy>=2.3) use Python 3.12 `type` statements, which mypy +# refuses to parse under python_version = 3.10 -- and 3.10 is the floor this +# code has to run on, so it stays. Treat numpy as Any instead: skip it, and +# follow_imports_for_stubs makes the skip apply to its .pyi files too. +[mypy-numpy.*] +follow_imports = skip +follow_imports_for_stubs = True + +# orjson is optional (see requirements.txt): the modules that use it fall back +# to the stdlib when `import orjson` fails. Whether mypy sees its stubs would +# otherwise depend on whether it happens to be installed -- installed, the +# `orjson = None` fallback is a type error and the stdlib branch "unreachable"; +# not installed, silencing either is an unused ignore. Treat it as Any always. +[mypy-orjson.*] +follow_imports = skip +follow_imports_for_stubs = True diff --git a/scripts/check_types.py b/scripts/check_types.py new file mode 100644 index 00000000..76873d07 --- /dev/null +++ b/scripts/check_types.py @@ -0,0 +1,99 @@ +#!/usr/bin/env python3 +"""Type-check the modules listed in mypy-clean.txt (the mypy ratchet). + +Most of src/ still has mypy errors, so CI can't require a clean `mypy src`. +Instead mypy-clean.txt lists the modules that *are* clean, and this script +fails if any of them regresses. When you make another module clean, add it to +the list; nothing ever comes off it. + +Imports are followed silently: a listed module is checked against the types of +everything it imports, but errors inside those imported modules are not +reported, so a clean file isn't failed by an unlisted neighbour. + +Usage: + python scripts/check_types.py # check the listed modules + python scripts/check_types.py --list # print the list and exit + +Extra arguments after ``--`` are passed to mypy. +Exit status: 0 clean, 1 mypy errors, 2 a bad list (missing file, duplicate, +unsorted, or empty). +""" + +import argparse +import subprocess # nosec B404 - list-form argv only, no shell # nosemgrep +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parent.parent +LIST_FILE = REPO_ROOT / "mypy-clean.txt" + + +def read_list(path: Path = LIST_FILE) -> list: + """The listed paths, in file order, with comments and blank lines dropped.""" + entries = [] + for raw in path.read_text(encoding="utf-8").splitlines(): + line = raw.split("#", 1)[0].strip() + if line: + entries.append(line) + return entries + + +def list_problems(entries: list, root: Path = REPO_ROOT) -> list: + """Why the list can't be used as-is; empty when it is fine.""" + problems = [] + if not entries: + problems.append(f"{LIST_FILE.name} lists no modules") + seen = set() + for entry in entries: + if entry in seen: + problems.append(f"listed twice: {entry}") + seen.add(entry) + if "\\" in entry: + problems.append(f"use forward slashes: {entry}") + elif not (root / entry).is_file(): + problems.append(f"listed but not found (renamed or deleted? update the list): {entry}") + if entries != sorted(entries): + problems.append(f"{LIST_FILE.name} is not sorted") + return problems + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + parser.add_argument("--list", action="store_true", help="print the listed modules and exit") + parser.add_argument("mypy_args", nargs="*", help="extra mypy arguments (after --)") + args = parser.parse_args(argv) + + entries = read_list() + problems = list_problems(entries) + if problems: + for problem in problems: + print(f"check_types: {problem}", file=sys.stderr) + return 2 + if args.list: + print("\n".join(entries)) + return 0 + + cmd = [ + sys.executable, "-m", "mypy", + "--config-file", str(REPO_ROOT / "mypy.ini"), + "--follow-imports=silent", + *args.mypy_args, + *entries, + ] + print(f"check_types: mypy on {len(entries)} modules from {LIST_FILE.name}", flush=True) + # This interpreter's mypy, fixed flags, and paths from the checked-in list. + result = subprocess.run(cmd, cwd=REPO_ROOT) # nosec B603 - list-form argv, no shell # nosemgrep + if result.returncode > 1: # mypy itself failed (bad config, crash) + return result.returncode + if result.returncode != 0: + print( + "check_types: a module on the mypy ratchet has type errors. Fix them " + f"(annotation-only where possible) rather than taking it off {LIST_FILE.name}.", + file=sys.stderr, + ) + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/backup_manager.py b/src/backup_manager.py index 895ac886..5f61e3f9 100644 --- a/src/backup_manager.py +++ b/src/backup_manager.py @@ -481,7 +481,8 @@ def validate_backup(zip_path: Path) -> Tuple[bool, str, Dict[str, Any]]: ): detected.append("plugin_uploads") - plugins: List[Dict[str, Any]] = [] + # Whatever the archive's manifest holds; checked below. + plugins: Any = [] if PLUGINS_MANIFEST_NAME in names: try: plugins = json.loads(zf.read(PLUGINS_MANIFEST_NAME).decode("utf-8")) diff --git a/src/base_odds_manager.py b/src/base_odds_manager.py index 095ce276..e1fbef08 100644 --- a/src/base_odds_manager.py +++ b/src/base_odds_manager.py @@ -16,7 +16,7 @@ import time import requests import json -from typing import Dict, Any, Optional, List +from typing import Dict, Any, Optional, List, cast from src.common.api_helper import DEFAULT_HTTP_HEADERS @@ -105,7 +105,7 @@ class BaseOddsManager: _FAILURE_COOLDOWN = 60.0 def get_odds(self, sport: str | None, league: str | None, event_id: str, - update_interval_seconds: int = None) -> Optional[Dict[str, Any]]: + update_interval_seconds: Optional[int] = None) -> Optional[Dict[str, Any]]: """ Fetch odds data for a specific game. @@ -126,7 +126,7 @@ class BaseOddsManager: cache_key = f"odds_espn_{sport}_{league}_{event_id}" # Check cache first - cached_data = self.cache_manager.get_with_auto_strategy(cache_key) + cached_data: Optional[Dict[str, Any]] = self.cache_manager.get_with_auto_strategy(cache_key) # Per-game chatter, logged on every update of every game on the # slate: debug, not the journal. @@ -204,7 +204,7 @@ class BaseOddsManager: cache_key, e, self._FAILURE_COOLDOWN) cached = self.cache_manager.get_with_auto_strategy(cache_key) - return None if _is_no_odds_marker(cached) else cached + return None if _is_no_odds_marker(cached) else cast(Optional[Dict[str, Any]], cached) def _extract_espn_data(self, data: Dict[str, Any]) -> Optional[Dict[str, Any]]: """ diff --git a/src/cache/memory_cache.py b/src/cache/memory_cache.py index 55ca010d..4c4783f1 100644 --- a/src/cache/memory_cache.py +++ b/src/cache/memory_cache.py @@ -8,7 +8,7 @@ import os import time import threading import logging -from typing import Dict, Any, Optional +from typing import Dict, Any, Optional, Union # Historical fixed ceiling, kept as the fallback when RAM cannot be read. DEFAULT_MAX_SIZE = 1000 @@ -70,13 +70,15 @@ class MemoryCache: """ self.logger = logging.getLogger(__name__) self._cache: Dict[str, Dict[str, Any]] = {} - self._timestamps: Dict[str, float] = {} + # Values are time.time() floats; get()/cleanup also accept a numeric + # string, as a timestamp may have been restored from serialized data. + self._timestamps: Dict[str, Union[float, str]] = {} self._lock = threading.Lock() self._max_size = max_size self._cleanup_interval = cleanup_interval self._last_cleanup = time.time() - def get(self, key: str, max_age: Optional[int] = None) -> Optional[Dict[str, Any]]: + def get(self, key: str, max_age: Optional[float] = None) -> Optional[Dict[str, Any]]: """ Get value from memory cache. @@ -200,6 +202,7 @@ class MemoryCache: max_age_for_cleanup = 3600 # 1 hour expired_keys = [] + timestamp: Optional[Union[float, str]] for key, timestamp in list(self._timestamps.items()): if isinstance(timestamp, str): try: diff --git a/src/common/api_helper.py b/src/common/api_helper.py index f3702639..b5f07276 100644 --- a/src/common/api_helper.py +++ b/src/common/api_helper.py @@ -11,12 +11,16 @@ import time from datetime import datetime from types import MappingProxyType from src.common.espn_dates import ESPN_MAX_LIMIT -from typing import Any, Dict, Mapping, Optional +from typing import TYPE_CHECKING, Any, Dict, Mapping, Optional, cast import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry +if TYPE_CHECKING: + # What Session() puts in .headers; the stubs only promise a MutableMapping. + from requests.structures import CaseInsensitiveDict + #: The User-Agent core sends to ESPN and other data APIs. It names the client #: and links to it: around 2026-08-04 ESPN began 403ing bare custom tokens @@ -84,7 +88,7 @@ class APIHelper: self.session.headers.update({**DEFAULT_HTTP_HEADERS, 'Connection': 'keep-alive'}) # Rate limiting - self._last_request_time = 0 # wall clock, reported by get_request_stats() + self._last_request_time: float = 0 # wall clock, reported by get_request_stats() # The interval is measured on time.monotonic(): a wall-clock step # back (NTP correcting a Pi with no RTC) made time_since_last # negative and the "remaining interval" sleep as long as the step. @@ -112,14 +116,14 @@ class APIHelper: cached = self._get_from_cache(cache_key, cache_ttl) if cached is not None: self.logger.debug(f"Using cached response for {cache_key}") - return cached + return cast(Dict[Any, Any], cached) # Rate limiting self._enforce_rate_limit() try: # Prepare request - request_headers = self.session.headers.copy() + request_headers = cast('CaseInsensitiveDict[Any]', self.session.headers).copy() if headers: request_headers.update(headers) @@ -133,7 +137,7 @@ class APIHelper: response.raise_for_status() # Parse JSON response - data = response.json() + data: Dict[Any, Any] = response.json() # Cache response if cache key provided if cache_key and self.cache_manager: @@ -247,7 +251,7 @@ class APIHelper: self._enforce_rate_limit() try: - request_headers = self.session.headers.copy() + request_headers = cast('CaseInsensitiveDict[Any]', self.session.headers).copy() if headers: request_headers.update(headers) @@ -260,7 +264,7 @@ class APIHelper: ) response.raise_for_status() - return response.json() + return cast(Optional[Dict[Any, Any]], response.json()) except requests.exceptions.RequestException as e: self.logger.error(f"POST request failed for {url}: {e}") diff --git a/src/common/espn_dates.py b/src/common/espn_dates.py index bda7e86a..ac517677 100644 --- a/src/common/espn_dates.py +++ b/src/common/espn_dates.py @@ -37,7 +37,7 @@ import time from concurrent.futures import ThreadPoolExecutor from datetime import date, timedelta from functools import partial -from typing import Any, Dict, List, Optional, Tuple +from typing import Any, Dict, List, Optional, Tuple, cast try: from src.common.json_body import response_json @@ -159,7 +159,7 @@ def espn_date_chunks(start: date, end: date) -> List[str]: return chunks -def merge_scoreboard_payloads(payloads: List[Dict[str, Any]]) -> Dict[str, Any]: +def merge_scoreboard_payloads(payloads: List[Any]) -> Dict[str, Any]: """Fold chunk responses into one scoreboard payload. Events are de-duplicated by id and keep first-seen order. Non-event keys @@ -202,7 +202,7 @@ def _fetch_one_chunk( timeout=timeout, ) response.raise_for_status() - return response_json(response) + return cast(Optional[Dict[str, Any]], response_json(response)) except Exception as exc: # noqa: BLE001 - see docstring if logger: logger.warning("ESPN chunk %s failed, skipping it: %s", chunk, exc) @@ -379,4 +379,4 @@ def fetch_espn_scoreboard( if data is not None: return data response.raise_for_status() - return response_json(response) + return cast(Dict[str, Any], response_json(response)) diff --git a/src/common/frame_timing.py b/src/common/frame_timing.py index ee554398..46305bd6 100644 --- a/src/common/frame_timing.py +++ b/src/common/frame_timing.py @@ -93,7 +93,7 @@ import tempfile import threading import time import traceback -from typing import Any, Callable, Dict, List, Optional, Tuple +from typing import Any, Callable, Dict, List, Optional, Tuple, TypedDict logger = logging.getLogger(__name__) @@ -483,7 +483,13 @@ class FrameTimingRecorder: raise -def watchdog_settings() -> Dict[str, float]: +class _WatchdogSettings(TypedDict, total=False): + """The StallWatchdog keyword arguments watchdog_settings() may set.""" + threshold: float + poll: float + + +def watchdog_settings() -> _WatchdogSettings: """StallWatchdog arguments from ``LEDMATRIX_STALL_WATCHDOG_MS``, if set. The poll comes down with the threshold, or a stall shorter than one poll diff --git a/src/common/logo_helper.py b/src/common/logo_helper.py index cc2f08fd..b23a762b 100644 --- a/src/common/logo_helper.py +++ b/src/common/logo_helper.py @@ -154,7 +154,7 @@ class LogoHelper: return None # Load image - logo = Image.open(logo_path) + logo: Image.Image = Image.open(logo_path) if logo.mode != 'RGBA': logo = logo.convert('RGBA') @@ -359,7 +359,7 @@ class LogoHelper: self._download_failures.clear() self.logger.debug("Logo cache cleared") - def get_cache_stats(self) -> Dict[str, int]: + def get_cache_stats(self) -> Dict[str, float]: """ Get cache statistics. diff --git a/src/common/render_gate.py b/src/common/render_gate.py index 5b7d5997..64dac717 100644 --- a/src/common/render_gate.py +++ b/src/common/render_gate.py @@ -46,7 +46,7 @@ import sys import threading import time from collections import deque -from typing import Any, Callable, Deque, List, Optional +from typing import Any, Callable, Deque, List, Optional, cast from src.common.frame_timing import binding_releases_gil @@ -103,8 +103,8 @@ def _held(lock: Any) -> bool: """Is ``lock`` held? RLocks report this thread's ownership; plain locks, anyone's.""" is_owned = getattr(lock, "_is_owned", None) if is_owned is not None: - return is_owned() - return lock.locked() + return cast(bool, is_owned()) + return cast(bool, lock.locked()) class RenderGate: diff --git a/src/common/scroll_config.py b/src/common/scroll_config.py index 67cb8a9f..12cfb73e 100644 --- a/src/common/scroll_config.py +++ b/src/common/scroll_config.py @@ -49,7 +49,7 @@ from __future__ import annotations import logging from dataclasses import dataclass, replace -from typing import Any, Dict, Optional +from typing import Any, Dict, List, Optional, cast from src.matrix_support import DEFAULT_REFRESH_LIMIT_HZ @@ -132,14 +132,14 @@ def crisp_ladder( refresh_hz: float = DEFAULT_REFRESH_HZ, max_frame_hold: int = MAX_FRAME_HOLD, max_pixels_per_frame: int = MAX_PIXELS_PER_FRAME, -): +) -> List[CrispSpeed]: """Every whole-pixel speed this panel can show, slowest first. Duplicates are collapsed keeping the gentlest option: 100 px/s is reachable as 1px every refresh or 2px every 2nd refresh, and the former moves in smaller increments, so that is the one worth offering. """ - best = {} + best: Dict[float, CrispSpeed] = {} for hold in range(1, max_frame_hold + 1): for ppf in range(1, max_pixels_per_frame + 1): pps = refresh_hz / hold * ppf @@ -434,7 +434,8 @@ def configure( # they start scrolling. configure() only reports what is needed. if choice: - requested = settings.requested_pixels_per_second + # Set whenever there is a crisp choice (see the replace() above). + requested = cast(float, settings.requested_pixels_per_second) if abs(requested - applied) > 0.05: log.info( "Scroll configured: %s (asked for %.1f px/s from %s; " diff --git a/src/common/sports_card.py b/src/common/sports_card.py index 3389f090..0c3b7831 100644 --- a/src/common/sports_card.py +++ b/src/common/sports_card.py @@ -144,7 +144,8 @@ def resolve_font_color(config: Optional[Dict[str, Any]], if len(matches) > 1: configured = [] for element in matches: - colour = element_color(config, element, None, mode) + # None as the default makes it come back when unconfigured. + colour = element_color(config, element, None, mode) # type: ignore[arg-type] if colour is not None and colour not in configured: configured.append(colour) if len(configured) == 1: diff --git a/src/config_service.py b/src/config_service.py index a062ba11..31822776 100644 --- a/src/config_service.py +++ b/src/config_service.py @@ -206,7 +206,8 @@ class ConfigService: # Sleep with periodic checks for stop signal for _ in range(int(self._watch_interval)): if self._stop_watching: - break + # Set from another thread; mypy keeps the while's narrowing. + break # type: ignore[unreachable] time.sleep(1) except Exception as e: diff --git a/src/deprecation.py b/src/deprecation.py index e0f09e02..4dd966d5 100644 --- a/src/deprecation.py +++ b/src/deprecation.py @@ -41,7 +41,7 @@ def deprecated(removal: str, alternative: Optional[str] = None) -> Callable[[F], warnings.warn(message, DeprecationWarning, stacklevel=2) return func(*args, **kwargs) - wrapper.__deprecated__ = message + wrapper.__deprecated__ = message # type: ignore[attr-defined] # functools' _Wrapped doesn't declare it return wrapper # type: ignore[return-value] return decorate diff --git a/src/device_location.py b/src/device_location.py index 1742b94c..8d4b97d2 100644 --- a/src/device_location.py +++ b/src/device_location.py @@ -24,7 +24,7 @@ next render. A location saved on the app itself always wins. import json import logging import time -from typing import Any, Callable, Dict, Iterable, List, Optional +from typing import Any, Callable, Dict, Iterable, List, Optional, cast GEOCODE_URL = "https://geocoding-api.open-meteo.com/v1/search" GEOCODE_TIMEOUT = 10 @@ -112,6 +112,7 @@ def parse_location(value: Any) -> Optional[Dict[str, Any]]: ``{"timezone": ...}`` when only the timezone box is filled) all mean the user has not given the app a place. """ + loc: Any if isinstance(value, dict): loc = value elif isinstance(value, str) and value.strip(): @@ -162,10 +163,10 @@ def geocode(city: str, state: Any = None, country: Any = None, """Look the city up on Open-Meteo. Raises on a network/HTTP failure.""" import requests - response = requests.get(GEOCODE_URL, params={ + response = requests.get(GEOCODE_URL, params=cast(Dict[str, Any], { "name": city, "count": GEOCODE_RESULT_COUNT, "language": "en", "format": "json", - }, timeout=timeout) + }), timeout=timeout) response.raise_for_status() best = pick_geocode_result(response.json().get("results") or [], state, country) if best is None: diff --git a/src/display_geometry.py b/src/display_geometry.py index b5655c12..ff3f71d1 100644 --- a/src/display_geometry.py +++ b/src/display_geometry.py @@ -132,6 +132,7 @@ def apply_pixel_mappers(width: int, height: int, mapper_config: str, ``multiplexing`` isn't modelled: its mappers give back the configured size for the panel sizes they are made for. """ + param: Optional[str] for entry in (mapper_config or '').split(';'): name, colon, param = entry.partition(':') name = name.lower() diff --git a/src/display_manager.py b/src/display_manager.py index 3b64abf8..7086b7ba 100644 --- a/src/display_manager.py +++ b/src/display_manager.py @@ -51,13 +51,16 @@ from src.pi5_matrix_support import is_raspberry_pi_5 import threading import time from collections import OrderedDict, deque -from typing import Dict, Any, List, Optional, Tuple +from typing import Dict, Any, List, Optional, Tuple, TYPE_CHECKING import math import zlib import freetype from src.common import snapshot_policy from src.common.frame_timing import FrameTimingRecorder + +if TYPE_CHECKING: + from src.common.render_gate import RenderGate from src.deprecation import deprecated from src.logging_config import get_logger from src.common.permission_utils import ( @@ -341,7 +344,7 @@ class DisplayManager: # A src.common.render_gate.RenderGate while Vegas runs with # vegas_scroll.prefetch_gate on: opened around each swap so the # prefetch thread only runs Python while this thread waits on vsync. - self.render_gate = None + self.render_gate: Optional['RenderGate'] = None # Timing of every presented frame, whoever drew it, for # scripts/frame_soak.py. See src/common/frame_timing.py. diff --git a/src/dynamic_team_resolver.py b/src/dynamic_team_resolver.py index 638ce339..f7a06213 100644 --- a/src/dynamic_team_resolver.py +++ b/src/dynamic_team_resolver.py @@ -20,7 +20,7 @@ Usage: import logging import time import requests -from typing import Dict, List +from typing import Any, Dict, List from src.common.api_helper import DEFAULT_HTTP_HEADERS @@ -35,7 +35,7 @@ class DynamicTeamResolver: """ # Cache for rankings data - _rankings_cache: Dict[str, List[str]] = {} + _rankings_cache: Dict[str, int] = {} # team abbreviation -> AP rank _cache_timestamp: float = 0 _cache_duration: int = 3600 # 1 hour cache # A failed or empty fetch is remembered briefly too: during an ESPN @@ -45,7 +45,7 @@ class DynamicTeamResolver: _failure_backoff: int = 300 # 5 minutes # Supported dynamic team patterns - DYNAMIC_PATTERNS = { + DYNAMIC_PATTERNS: Dict[str, Dict[str, Any]] = { 'AP_TOP_25': {'sport': 'ncaa_fb', 'limit': 25}, 'AP_TOP_10': {'sport': 'ncaa_fb', 'limit': 10}, 'AP_TOP_5': {'sport': 'ncaa_fb', 'limit': 5}, diff --git a/src/exceptions.py b/src/exceptions.py index 15aa5e69..59ab196c 100644 --- a/src/exceptions.py +++ b/src/exceptions.py @@ -5,11 +5,13 @@ Provides specific exception types for different error categories, enabling better error handling and debugging. """ +from typing import Optional + class LEDMatrixError(Exception): """Base exception for all LEDMatrix errors.""" - def __init__(self, message: str, context: dict = None): + def __init__(self, message: str, context: Optional[dict] = None): """ Initialize the exception. @@ -32,7 +34,7 @@ class LEDMatrixError(Exception): class CacheError(LEDMatrixError): """Exception raised for cache-related errors.""" - def __init__(self, message: str, cache_key: str = None, context: dict = None): + def __init__(self, message: str, cache_key: Optional[str] = None, context: Optional[dict] = None): """ Initialize cache error. @@ -53,7 +55,7 @@ class CacheError(LEDMatrixError): class ConfigError(LEDMatrixError): """Exception raised for configuration-related errors.""" - def __init__(self, message: str, config_path: str = None, field: str = None, context: dict = None): + def __init__(self, message: str, config_path: Optional[str] = None, field: Optional[str] = None, context: Optional[dict] = None): """ Initialize config error. @@ -79,7 +81,7 @@ class ConfigError(LEDMatrixError): class PluginError(LEDMatrixError): """Exception raised for plugin-related errors.""" - def __init__(self, message: str, plugin_id: str = None, context: dict = None): + def __init__(self, message: str, plugin_id: Optional[str] = None, context: Optional[dict] = None): """ Initialize plugin error. @@ -100,7 +102,7 @@ class PluginError(LEDMatrixError): class DisplayError(LEDMatrixError): """Exception raised for display-related errors.""" - def __init__(self, message: str, display_mode: str = None, context: dict = None): + def __init__(self, message: str, display_mode: Optional[str] = None, context: Optional[dict] = None): """ Initialize display error. diff --git a/src/logging_config.py b/src/logging_config.py index 23af6bcf..d57591c8 100644 --- a/src/logging_config.py +++ b/src/logging_config.py @@ -125,6 +125,7 @@ def setup_logging( root_logger.handlers.clear() # Create formatter based on type + formatter: logging.Formatter if format_type == 'json': formatter = StructuredFormatter() else: @@ -244,7 +245,7 @@ class PluginLoggerAdapter(logging.LoggerAdapter): def process(self, msg, kwargs): extra = dict(kwargs.get('extra') or {}) - extra.setdefault('plugin_id', self.extra.get('plugin_id')) + extra.setdefault('plugin_id', self.extra.get('plugin_id')) # type: ignore[union-attr] # get_logger always passes a dict kwargs['extra'] = extra return msg, kwargs @@ -290,7 +291,7 @@ def log_with_context( operation_id: Optional operation ID for request tracking exc_info: Optional exception info for error logging """ - extra = {} + extra: Dict[str, Any] = {} if context: extra['context'] = context diff --git a/src/logo_downloader.py b/src/logo_downloader.py index b7fa239a..b4f3f64e 100644 --- a/src/logo_downloader.py +++ b/src/logo_downloader.py @@ -13,7 +13,7 @@ import time import logging import requests import json -from typing import Dict, List, Optional, Tuple +from typing import Dict, List, Optional, Tuple, Union from pathlib import Path from PIL import Image, ImageDraw, ImageFont, UnidentifiedImageError from src.common.font_layout import load_truetype, resolve_asset_path @@ -481,7 +481,7 @@ class LogoDownloader: logger.info(f"Fetching team data for {league} from ESPN API...") response = self.session.get(api_url, params={'limit':1000},headers=self.headers, timeout=self.request_timeout) response.raise_for_status() - data = response.json() + data: Dict = response.json() logger.info(f"Successfully fetched team data for {league}") return data @@ -505,7 +505,7 @@ class LogoDownloader: logger.info(f"Fetching team data for team {team_id} in {league} from ESPN API...") response = self.session.get(f"{api_url}/{team_id}", headers=self.headers, timeout=self.request_timeout) response.raise_for_status() - data = response.json() + data: Dict = response.json() logger.info(f"Successfully fetched team data for {team_id} in {league}") return data @@ -815,6 +815,7 @@ class LogoDownloader: draw = ImageDraw.Draw(logo) # Try to load a font, fallback to default + font: Optional[Union[ImageFont.FreeTypeFont, ImageFont.ImageFont]] try: font = load_truetype(resolve_asset_path("assets/fonts/PressStart2P-Regular.ttf"), 12) except (OSError, IOError): diff --git a/src/plugin_system/compatibility.py b/src/plugin_system/compatibility.py index 28a5a4f2..b3687222 100644 --- a/src/plugin_system/compatibility.py +++ b/src/plugin_system/compatibility.py @@ -191,7 +191,7 @@ def satisfies_compatible_versions( return any(parsed) -def declared_min_version(manifest: Dict[str, Any]) -> Optional[str]: +def declared_min_version(manifest: Dict[str, Any]) -> Any: """The core version this plugin says it needs, or ``None`` if it doesn't say. Checked in order of specificity. `ledmatrix_min` is the deprecated spelling @@ -206,6 +206,10 @@ def declared_min_version(manifest: Dict[str, Any]) -> Optional[str]: untrustworthy-core branch of :func:`check` calls this for *every* manifest, so one malformed file would take down the install path rather than just itself. A shape we do not recognise means "no declared floor". + + The value is returned as the manifest holds it -- normally a version + string, but nothing here checks that; callers hand it to + :func:`parse_semver`, which accepts anything. """ declared = manifest.get('min_ledmatrix_version') if not declared: @@ -222,7 +226,7 @@ def declared_min_version(manifest: Dict[str, Any]) -> Optional[str]: return None -def is_update_available(installed_version: str, latest_version: str) -> bool: +def is_update_available(installed_version: Any, latest_version: Any) -> bool: """Return True when the registry's ``latest_version`` is strictly newer than the installed version. diff --git a/src/plugin_system/plugin_dirs.py b/src/plugin_system/plugin_dirs.py index 46837b58..a28f8104 100644 --- a/src/plugin_system/plugin_dirs.py +++ b/src/plugin_system/plugin_dirs.py @@ -45,7 +45,7 @@ from __future__ import annotations import json from dataclasses import dataclass, field from pathlib import Path -from typing import Any, Dict, Iterable, List, Optional, Set, Union +from typing import Any, Dict, Iterable, List, Optional, Set, Union, cast from src.common.path_safety import safe_path_component @@ -91,7 +91,7 @@ class PluginDirEntry: """One candidate directory and its manifest, read once.""" path: Path status: str - manifest: Optional[Any] = None + manifest: Any = None error: Optional[BaseException] = None @property @@ -103,7 +103,8 @@ class PluginDirEntry: """The manifest's ``id`` when the manifest is usable, else None.""" if self.status != ManifestStatus.OK: return None - return self.manifest['id'] + # OK means a dict whose "id" is a non-empty string (_read_entry). + return cast(str, self.manifest['id']) @property def manifest_parses(self) -> bool: @@ -240,7 +241,7 @@ class PluginDirectoryIndex: # -- lookup ----------------------------------------------------------- - def find(self, plugin_id: str, *, prefix: bool, case_insensitive: bool, + def find(self, plugin_id: Any, *, prefix: bool, case_insensitive: bool, by_manifest: bool = True) -> Optional[Path]: """Resolve ``plugin_id`` within this directory (rules in the module doc).""" plugin_id = _lookup_id(plugin_id) @@ -270,7 +271,7 @@ def _lookup_id(plugin_id: Any) -> Optional[str]: plugin_id = safe_path_component(plugin_id) if plugin_id is None or is_ignored_dir_name(plugin_id): return None - return plugin_id + return cast(str, plugin_id) # safe_path_component returned a str def _candidate_names(plugin_id: str, prefix: bool) -> List[str]: diff --git a/src/plugin_system/plugin_executor.py b/src/plugin_system/plugin_executor.py index a2014145..7fea423c 100644 --- a/src/plugin_system/plugin_executor.py +++ b/src/plugin_system/plugin_executor.py @@ -6,7 +6,7 @@ error isolation, and performance monitoring. """ import time -from typing import Any, Optional, Callable +from typing import Any, Dict, Optional, Callable from threading import Thread import logging @@ -62,7 +62,7 @@ class PluginExecutor: plugin_context = f"plugin {plugin_id}" if plugin_id else "plugin" # Use threading-based timeout (more reliable than signal-based) - result_container = {'value': None, 'exception': None, 'completed': False} + result_container: Dict[str, Any] = {'value': None, 'exception': None, 'completed': False} def target(): try: diff --git a/src/plugin_system/plugin_loader.py b/src/plugin_system/plugin_loader.py index 3053387d..b5286cd8 100644 --- a/src/plugin_system/plugin_loader.py +++ b/src/plugin_system/plugin_loader.py @@ -277,6 +277,7 @@ class PluginLoader: Path to plugin directory or None if not found. An id that is not one plain path segment finds nothing. """ + plugin_dir: Optional[Path] # Strategy 1: Use mapping from discovery if plugin_directories and plugin_id in plugin_directories: plugin_dir = plugin_directories[plugin_id] diff --git a/src/plugin_system/resource_monitor.py b/src/plugin_system/resource_monitor.py index adeb6ed6..01b3204e 100644 --- a/src/plugin_system/resource_monitor.py +++ b/src/plugin_system/resource_monitor.py @@ -8,7 +8,7 @@ Provides resource limits and performance monitoring. import math import time import threading -from typing import Dict, Optional, Any, Callable +from typing import Dict, Optional, Any, Callable, cast from dataclasses import dataclass, field, fields from src.logging_config import get_logger @@ -209,7 +209,7 @@ class PluginResourceMonitor: # (not \"int\") to str"). Coerce here, where there is still a cache # key to name in the warning. declared = {f.name: f.type for f in fields(ResourceMetrics)} - usable = {} + usable: Dict[str, Any] = {} for key, value in cached.items(): if key not in known: continue @@ -299,7 +299,7 @@ class PluginResourceMonitor: if not self.enable_monitoring or self._process is None: return 0.0 try: - return self._process.memory_info().rss / 1024 / 1024 + return cast(float, self._process.memory_info().rss / 1024 / 1024) except Exception: return 0.0 @@ -313,7 +313,7 @@ class PluginResourceMonitor: if not self.enable_monitoring or self._process is None: return 0.0 try: - return self._process.cpu_percent(interval=None) + return cast(float, self._process.cpu_percent(interval=None)) except Exception: return 0.0 diff --git a/src/plugin_system/saved_repositories.py b/src/plugin_system/saved_repositories.py index 86dcd250..11845f43 100644 --- a/src/plugin_system/saved_repositories.py +++ b/src/plugin_system/saved_repositories.py @@ -7,7 +7,7 @@ Manages saved GitHub repository URLs for easy plugin discovery and installation. import json import os from pathlib import Path -from typing import List, Dict, Optional +from typing import List, Dict, Optional, cast from src.logging_config import get_logger from src.plugin_system.repo_urls import normalize_repo_url @@ -37,7 +37,7 @@ class SavedRepositoriesManager: if isinstance(data, list): return data elif isinstance(data, dict) and 'repositories' in data: - return data['repositories'] + return cast(List[Dict[str, str]], data['repositories']) else: return [] return [] diff --git a/src/plugin_system/state_reconciliation.py b/src/plugin_system/state_reconciliation.py index 865b8425..13a23123 100644 --- a/src/plugin_system/state_reconciliation.py +++ b/src/plugin_system/state_reconciliation.py @@ -9,7 +9,7 @@ Detects and fixes inconsistencies between: """ import json -from typing import Dict, Any, List, Set +from typing import Dict, Any, List, Set, cast from dataclasses import dataclass from enum import Enum from pathlib import Path @@ -237,7 +237,7 @@ class StateReconciliation: state_manager_state = self._get_state_manager_state() # Find all unique plugin IDs - all_plugin_ids = set() + all_plugin_ids: Set[str] = set() all_plugin_ids.update(config_state.keys()) all_plugin_ids.update(disk_state.keys()) all_plugin_ids.update(manager_state.keys()) @@ -380,7 +380,7 @@ class StateReconciliation: state_manager_state: Dict[str, Dict[str, Any]] ) -> List[Inconsistency]: """Check consistency for a single plugin.""" - inconsistencies = [] + inconsistencies: List[Inconsistency] = [] if plugin_id in CORE_CONFIG_KEYS: # A plugin whose id is a core setting's key ('display', 'sync', @@ -496,7 +496,8 @@ class StateReconciliation: # Bring the state manager in sync with config rather than the reverse, # so that manual config edits (or the state left behind after an # uninstall+reinstall cycle) don't silently override the user's intent. - config_enabled = inconsistency.expected_state.get('enabled') + # Always set for this type (see _check_plugin_consistency). + config_enabled = cast(bool, inconsistency.expected_state.get('enabled')) success = self.state_manager.set_plugin_enabled(inconsistency.plugin_id, config_enabled) if success: self.logger.info( diff --git a/src/plugin_system/testing/loading.py b/src/plugin_system/testing/loading.py index 2a0dd8e0..beca27b9 100644 --- a/src/plugin_system/testing/loading.py +++ b/src/plugin_system/testing/loading.py @@ -7,7 +7,7 @@ plugin discovery / manifest / config-default logic lives in exactly one place. import json from pathlib import Path -from typing import Any, Dict, Optional, Sequence, Union +from typing import Any, Dict, Optional, Sequence, Union, cast def find_plugin_dir(plugin_id: str, search_dirs: Sequence[Union[str, Path]]) -> Optional[Path]: @@ -39,7 +39,7 @@ def load_manifest(plugin_dir: Union[str, Path]) -> Dict[str, Any]: if not manifest_path.exists(): raise FileNotFoundError(f"No manifest.json in {plugin_dir}") with open(manifest_path, 'r', encoding='utf-8') as f: - return json.load(f) + return cast(Dict[str, Any], json.load(f)) def merge_config(base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]: @@ -64,7 +64,7 @@ def load_schema(plugin_dir: Union[str, Path]) -> Optional[Dict[str, Any]]: if not schema_path.exists(): return None with open(schema_path, 'r', encoding='utf-8') as f: - return json.load(f) + return cast(Optional[Dict[str, Any]], json.load(f)) def load_config_defaults(plugin_dir: Union[str, Path]) -> Dict[str, Any]: @@ -124,7 +124,7 @@ def load_harness_spec(plugin_dir: Union[str, Path]) -> Dict[str, Any]: if not spec_path.exists(): return {} with open(spec_path, 'r', encoding='utf-8') as f: - spec = json.load(f) + spec: Dict[str, Any] = json.load(f) # Resolve mock_data path and inline its contents for convenience. mock_rel = spec.get('mock_data') diff --git a/src/plugin_system/testing/mocks.py b/src/plugin_system/testing/mocks.py index a103670b..c3c9bc31 100644 --- a/src/plugin_system/testing/mocks.py +++ b/src/plugin_system/testing/mocks.py @@ -6,7 +6,7 @@ and plugin_manager for use in plugin unit tests. """ import warnings -from typing import Dict, Any, Optional +from typing import Dict, Any, List, Optional from PIL import Image #: Why draw_image() warns. Kept (rather than removed) so existing plugin test @@ -32,7 +32,7 @@ class MockDisplayManager: self.image = Image.new('RGB', (width, height), color=(0, 0, 0)) self.clear_called = False self.update_called = False - self.draw_calls = [] + self.draw_calls: List[Dict[str, Any]] = [] def clear(self): """Clear the display.""" @@ -43,7 +43,7 @@ class MockDisplayManager: """Update the display.""" self.update_called = True - def draw_text(self, text: str, x: int = None, y: int = None, color: tuple = (255, 255, 255), + def draw_text(self, text: str, x: Optional[int] = None, y: Optional[int] = None, color: tuple = (255, 255, 255), font=None, small_font: bool = False, centered: bool = False): """Draw text on the display. @@ -166,8 +166,8 @@ class MockConfigManager: def __init__(self, config: Optional[Dict[str, Any]] = None): self._config = config or {} - self.load_config_calls = [] - self.save_config_calls = [] + self.load_config_calls: List[Dict[str, Any]] = [] + self.save_config_calls: List[Dict[str, Any]] = [] def load_config(self) -> Dict[str, Any]: """Load configuration.""" diff --git a/src/vegas_mode/coordinator.py b/src/vegas_mode/coordinator.py index 06d7b53f..3789ec03 100644 --- a/src/vegas_mode/coordinator.py +++ b/src/vegas_mode/coordinator.py @@ -83,6 +83,8 @@ class VegasModeCoordinator: # Class-level so coordinators built without __init__ (tests) have it. _last_live_check: float = float('-inf') + # Set only while Vegas has changed the GIL switch interval; read with getattr. + _saved_switch_interval: Optional[float] def __init__( self, @@ -331,7 +333,7 @@ class VegasModeCoordinator: self._saved_switch_interval = sys.getswitchinterval() sys.setswitchinterval(ms / 1000.0) logger.info("Vegas: GIL switch interval %.1fms (was %.1fms)", - ms, self._saved_switch_interval * 1000.0) + ms, self._saved_switch_interval * 1000.0) # type: ignore[operator] # set just above; getattr hides it def _restore_switch_interval(self) -> None: saved = getattr(self, '_saved_switch_interval', None) @@ -842,7 +844,7 @@ class VegasModeCoordinator: plugin_id = self.render_pipeline.next_static_trigger() if not plugin_id: return None - plugin = self.plugin_manager.get_plugin(plugin_id) + plugin: Optional['BasePlugin'] = self.plugin_manager.get_plugin(plugin_id) if not plugin: logger.debug("[%s] STATIC turn reached, but the plugin is no longer loaded", plugin_id) diff --git a/src/vegas_mode/stream_manager.py b/src/vegas_mode/stream_manager.py index f02da6b3..1c4b2d41 100644 --- a/src/vegas_mode/stream_manager.py +++ b/src/vegas_mode/stream_manager.py @@ -14,7 +14,7 @@ Supports three display modes: import logging import threading import time -from typing import Optional, List, Dict, Any, Deque, Tuple, TYPE_CHECKING +from typing import Optional, List, Dict, Any, Deque, Tuple, TYPE_CHECKING, cast from collections import deque from dataclasses import dataclass, field from PIL import Image @@ -698,7 +698,7 @@ class StreamManager: if plugin is None: return False try: - return plugin.get_vegas_display_mode() == VegasDisplayMode.STATIC + return cast(bool, plugin.get_vegas_display_mode() == VegasDisplayMode.STATIC) except Exception: logger.debug("[%s] get_vegas_display_mode() failed; treating as not STATIC", plugin_id, exc_info=True) diff --git a/src/web_interface/api_helpers.py b/src/web_interface/api_helpers.py index c3bba7f2..2008befb 100644 --- a/src/web_interface/api_helpers.py +++ b/src/web_interface/api_helpers.py @@ -109,7 +109,7 @@ def exception_error_response( ) -def validate_request_json(required_fields: list, data: Optional[Dict] = None) -> Tuple[Optional[Dict], Optional[Any]]: +def validate_request_json(required_fields: list, data: Any = None) -> Tuple[Optional[Dict], Optional[Any]]: """ Validate request JSON has required fields. diff --git a/src/web_interface/config_arrays.py b/src/web_interface/config_arrays.py index 87e3e95f..0e644a05 100644 --- a/src/web_interface/config_arrays.py +++ b/src/web_interface/config_arrays.py @@ -30,7 +30,7 @@ def _is_index_dict(value: Any) -> bool: return isinstance(value, dict) and all(str(k).isdigit() for k in value) -def coerce_array_shapes(config: Dict[str, Any], schema_props: Dict[str, Any], +def coerce_array_shapes(config: Any, schema_props: Dict[str, Any], short_lists_take_default: bool = False) -> None: """Turn position-keyed dicts into lists wherever the schema has an array. @@ -64,7 +64,7 @@ def coerce_array_shapes(config: Dict[str, Any], schema_props: Dict[str, Any], and len(value) < min_items and isinstance(default, list) and len(default) >= min_items): value = config[key] = list(default) - items_schema = prop_schema.get('items') + items_schema: Any = prop_schema.get('items') if (_schema_type_is(items_schema, 'object') and 'properties' in items_schema): for element in value: diff --git a/src/web_interface/error_handler.py b/src/web_interface/error_handler.py index 376636a2..c3360def 100644 --- a/src/web_interface/error_handler.py +++ b/src/web_interface/error_handler.py @@ -152,7 +152,7 @@ def create_success_response( Returns: Dictionary for jsonify """ - response = { + response: dict[str, Any] = { "status": "success" } diff --git a/src/web_interface/errors.py b/src/web_interface/errors.py index 2fe03995..4d3f9fd6 100644 --- a/src/web_interface/errors.py +++ b/src/web_interface/errors.py @@ -142,7 +142,7 @@ class WebInterfaceError: def to_dict(self) -> Dict[str, Any]: """Convert error to dictionary for JSON response.""" - result = { + result: Dict[str, Any] = { "status": "error", "error_code": self.error_code.value, "message": self.message, diff --git a/src/web_interface/secret_helpers.py b/src/web_interface/secret_helpers.py index 2680591c..f8dce61f 100644 --- a/src/web_interface/secret_helpers.py +++ b/src/web_interface/secret_helpers.py @@ -5,10 +5,10 @@ Provides functions for identifying, masking, separating, and filtering secret fields in plugin configurations based on JSON Schema x-secret markers. """ -from typing import Any, Dict, Optional, Set, Tuple +from typing import Any, Dict, Optional, Set, Tuple, cast -def find_secret_fields(properties: Dict[str, Any], prefix: str = '') -> Set[str]: +def find_secret_fields(properties: Any, prefix: str = '') -> Set[str]: """Find all fields marked with ``x-secret: true`` in a JSON Schema properties dict. Recurses into nested objects and array items to discover secrets at any @@ -336,4 +336,4 @@ def _contains_mask(value: Any) -> bool: return any(_contains_mask(v) for v in value.values()) if isinstance(value, list): return any(_contains_mask(item) for item in value) - return value == SECRET_MASK + return cast(bool, value == SECRET_MASK)