diff --git a/PyWebApp/cookieMonster.py b/PyWebApp/cookieMonster.py index 97fcb0a..88b11b0 100644 --- a/PyWebApp/cookieMonster.py +++ b/PyWebApp/cookieMonster.py @@ -7,7 +7,7 @@ from hashlib import sha256 from logging import error, warning from pickle import dumps, loads -from typing import Any, Optional +from typing import Any from cryptography.fernet import Fernet, InvalidToken @@ -141,7 +141,7 @@ def makeCookie(data: Any, key: bytes = b"") -> str: return "" -def eatCookie(cookie: str) -> Optional[Any]: +def eatCookie(cookie: str) -> Any | None: """Decrypt and deserialise a cookie string produced by makeCookie. Returns the original data on success. diff --git a/PyWebApp/index.py b/PyWebApp/index.py index 5339df6..9e13aed 100644 --- a/PyWebApp/index.py +++ b/PyWebApp/index.py @@ -2,7 +2,7 @@ import os import time from functools import wraps -from typing import Any, Dict, Optional +from typing import Any # Import the CherryPy modules. import cherrypy @@ -28,11 +28,11 @@ # looks for files in the templates directory -def render(filename: str, data: Dict) -> str: +def render(filename: str, data: dict) -> str: return django.template.loader.get_template(filename).render(data) -def _get_cookie() -> Optional[Any]: +def _get_cookie() -> Any | None: """Read and decode the app cookie from the current request. Returns None if the cookie is absent or invalid.""" morsel = cherrypy.request.cookie.get(myApp) diff --git a/checkmedia.py b/checkmedia.py index 6247977..3749e6f 100644 --- a/checkmedia.py +++ b/checkmedia.py @@ -10,7 +10,6 @@ import threading import time from sys import argv, exit -from typing import Optional tlock = threading.Lock() tqueue: queue.Queue[str] = queue.Queue() @@ -23,12 +22,12 @@ # Type alias for the 6-element DB row: (hash, atime, mtime, ctime, size, whent) # whent is time.struct_time on success or float (time.time()) on parse failure DBRow = tuple[ - Optional[str], - Optional[int], - Optional[int], - Optional[int], - Optional[int], - Optional[time.struct_time | float], + str | None, + int | None, + int | None, + int | None, + int | None, + time.struct_time | float | None, ] # Type alias for a pending batch insert row: (path, hash, atime, mtime, ctime, size) @@ -129,7 +128,7 @@ def getDB(cur: sqlite3.Cursor, path: str) -> DBRow: "SELECT hash, atime, mtime, ctime, size, whent FROM media WHERE path=? ORDER BY whent DESC LIMIT 1;", (path,), ) - results: Optional[tuple] = cur.fetchone() + results: tuple | None = cur.fetchone() except Exception as ex: logger("Unable to get data from media DB: %s" % str(ex)) return (None, None, None, None, None, None) @@ -195,7 +194,7 @@ def hashThreads(cur: sqlite3.Cursor, paths: list[str]) -> None: if fstat[1] == 0: continue - dbhash, atime, mtime, ctime, size, whent = getDB(cur, fullpath) + dbhash, _atime, mtime, ctime, size, _whent = getDB(cur, fullpath) if dbhash is None: fhash: str = computeHash(fullpath) diff --git a/cookieMonster.py b/cookieMonster.py index 6e75af1..b6b0d73 100644 --- a/cookieMonster.py +++ b/cookieMonster.py @@ -7,7 +7,7 @@ from hashlib import sha256 from logging import error, warning from pickle import dumps, loads -from typing import Any, Optional +from typing import Any from cryptography.fernet import Fernet, InvalidToken @@ -139,7 +139,7 @@ def makeCookie(data: Any, key: bytes = b"") -> str: return "" -def eatCookie(cookie: str) -> Optional[Any]: +def eatCookie(cookie: str) -> Any | None: """Decrypt and deserialise a cookie string produced by makeCookie. Returns the original data on success. diff --git a/heartbeat.py b/heartbeat.py index 231ab33..23db585 100644 --- a/heartbeat.py +++ b/heartbeat.py @@ -72,12 +72,12 @@ import sys import threading import time -from typing import Callable, Optional +from collections.abc import Callable log = logging.getLogger("heartbeat") -def _close_socket(sock: Optional[socket.socket]) -> None: +def _close_socket(sock: socket.socket | None) -> None: """Best-effort shutdown and close. Safe with None.""" if sock is None: return @@ -110,12 +110,12 @@ def __init__( my_ip: str, peer_ip: str, port: int = 53281, - peer_port: Optional[int] = None, + peer_port: int | None = None, *, managed_process: str = "my_server.py", - health_check: Optional[Callable[[], bool]] = None, + health_check: Callable[[], bool] | None = None, health_check_cache_seconds: float = 1.0, - on_state_change: Optional[Callable[[str, str], None]] = None, + on_state_change: Callable[[str, str], None] | None = None, heartbeat_interval: float = 1.0, heartbeat_timeout: float = 5.0, local_check_interval: float = 10.0, @@ -161,7 +161,7 @@ def __init__( # Cached health-check result. Forking ps on every heartbeat (which # can fire from multiple threads at once during a reconcile) adds # up; cache for `health_check_cache_seconds` to amortize. - self._health_cache_value: Optional[bool] = None + self._health_cache_value: bool | None = None self._health_cache_time: float = 0.0 self._health_cache_lock = threading.Lock() @@ -171,20 +171,20 @@ def __init__( self._peer_announced_departure = threading.Event() # Listener (server) side. - self._ssock: Optional[socket.socket] = None - self._listen_thread: Optional[threading.Thread] = None - self._reaper_thread: Optional[threading.Thread] = None + self._ssock: socket.socket | None = None + self._listen_thread: threading.Thread | None = None + self._reaper_thread: threading.Thread | None = None self._workers: list[socket.socket] = [] self._workers_lock = threading.Lock() self._worker_threads: list[threading.Thread] = [] self._worker_threads_lock = threading.Lock() # Outbound (client) side — our connection to the peer. - self._csock: Optional[socket.socket] = None + self._csock: socket.socket | None = None self._csock_lock = threading.Lock() - self._heartbeat_thread: Optional[threading.Thread] = None - self._last_successful_contact: Optional[float] = None - self._loop_start_time: Optional[float] = None + self._heartbeat_thread: threading.Thread | None = None + self._last_successful_contact: float | None = None + self._loop_start_time: float | None = None # ------------------------------------------------------------------ # # State accessors # @@ -233,6 +233,7 @@ def _do_health_check(self) -> bool: stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=5.0, + check=False, ) return bool(result.stdout.strip()) except Exception as ex: @@ -290,7 +291,7 @@ def _serve_connection(self, client_sock: socket.socket) -> None: while not self._shutdown.is_set(): try: cmd = client_sock.recv(1) - except socket.timeout: + except TimeoutError: continue except OSError as ex: log.debug("Peer connection error: %s", ex) @@ -335,7 +336,7 @@ def _accept_loop(self) -> None: while not self._shutdown.is_set(): try: client_sock, _addr = self._ssock.accept() - except socket.timeout: + except TimeoutError: continue except OSError: if self._shutdown.is_set(): @@ -376,9 +377,9 @@ def _reaper_loop(self) -> None: # Heartbeat sender (client side) # # ------------------------------------------------------------------ # - def _connect_locked(self) -> Optional[socket.socket]: + def _connect_locked(self) -> socket.socket | None: """Open a fresh client socket to the peer. Caller holds _csock_lock.""" - sock: Optional[socket.socket] = None + sock: socket.socket | None = None try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(self._heartbeat_timeout) @@ -612,7 +613,7 @@ def _heartbeat_loop(self) -> None: try: sock.sendall(self.PROTO_QUERY_STATE) reply = sock.recv(1) - except socket.timeout: + except TimeoutError: log.info( "No heartbeat reply within %.1fs", self._heartbeat_timeout, @@ -669,7 +670,7 @@ def start(self) -> None: if self._ssock is not None: raise RuntimeError("heartbeat already started") - ssock: Optional[socket.socket] = None + ssock: socket.socket | None = None try: ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) ssock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) @@ -699,7 +700,7 @@ def _notify_peer_goodbye(self) -> None: """Open a short-lived socket and send ``PROTO_GOODBYE``. Best effort — failure is logged at debug level and otherwise ignored (peer will fail over via the grace period instead).""" - sock: Optional[socket.socket] = None + sock: socket.socket | None = None try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(2.0) @@ -757,7 +758,7 @@ def stop(self) -> None: def install_signal_handlers( self, - signals: Optional[tuple[int, ...]] = None, + signals: tuple[int, ...] | None = None, install_dump: bool = True, ) -> None: """Install handlers that call ``stop()`` on the given signals. @@ -831,7 +832,7 @@ def dump_state(self) -> str: ] return "\n".join(lines) - def wait_for_shutdown(self, timeout: Optional[float] = None) -> bool: + def wait_for_shutdown(self, timeout: float | None = None) -> bool: return self._shutdown.wait(timeout=timeout) @@ -840,7 +841,7 @@ def wait_for_shutdown(self, timeout: Optional[float] = None) -> bool: # --------------------------------------------------------------------------- # -def _parse_args(argv: Optional[list[str]] = None) -> argparse.Namespace: +def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(prog="heartbeat") parser.add_argument( "-m", @@ -875,7 +876,7 @@ def _parse_args(argv: Optional[list[str]] = None) -> argparse.Namespace: return parser.parse_args(argv) -def main(argv: Optional[list[str]] = None) -> int: +def main(argv: list[str] | None = None) -> int: args = _parse_args(argv) logging.basicConfig( format="%(asctime)s %(levelname)s\t%(message)s", diff --git a/linux_sys_stats.py b/linux_sys_stats.py index 0386574..56e04ea 100644 --- a/linux_sys_stats.py +++ b/linux_sys_stats.py @@ -4,10 +4,9 @@ import subprocess import sys from platform import node -from typing import Dict, List -def get_top() -> List[str]: +def get_top() -> list[str]: try: top = subprocess.Popen( ["/usr/bin/top", "-b", "-n", "1", "-w", "240"], @@ -16,7 +15,7 @@ def get_top() -> List[str]: stderr=subprocess.PIPE, ) lines = top.stdout.readlines() - retList: List[str] = [] + retList: list[str] = [] for line in lines: retList.append(line.decode("utf-8").strip()) return retList @@ -25,8 +24,8 @@ def get_top() -> List[str]: sys.exit(1) -def parse_top(top: List[str]) -> Dict: - retDict: Dict = {"hostname": node()} +def parse_top(top: list[str]) -> dict: + retDict: dict = {"hostname": node()} for line in top: if not line: continue # skip blank lines @@ -95,7 +94,7 @@ def parse_top(top: List[str]) -> Dict: "available": avail, } elif fields[0] == "PID": - out_fields: Dict[str, int] = {} + out_fields: dict[str, int] = {} for lnum, name in enumerate(fields): out_fields[name] = lnum else: diff --git a/mac_sys_stats.py b/mac_sys_stats.py index c77264e..05a7291 100755 --- a/mac_sys_stats.py +++ b/mac_sys_stats.py @@ -5,10 +5,9 @@ import sys import time from platform import node -from typing import Dict, List -def get_top() -> List[str]: +def get_top() -> list[str]: try: top = subprocess.Popen( ["/usr/bin/top", "-l", "1"], @@ -18,7 +17,7 @@ def get_top() -> List[str]: ) lines = top.stdout.readlines() top.terminate() - retList: List[str] = [] + retList: list[str] = [] for line in lines: retList.append(line.decode("utf-8").strip()) return retList @@ -27,9 +26,9 @@ def get_top() -> List[str]: sys.exit(1) -def parse_top(top: List[str]) -> Dict: +def parse_top(top: list[str]) -> dict: year = time.localtime().tm_year - retDict: Dict = {"hostname": node()} + retDict: dict = {"hostname": node()} for line in top: if not line: continue # skip blank lines @@ -134,7 +133,7 @@ def parse_top(top: List[str]) -> Dict: "data_written": memout, } elif line.startswith("PID"): - out_fields: Dict[str, int] = {} + out_fields: dict[str, int] = {} for lnum, name in enumerate(fields): out_fields[name] = lnum else: diff --git a/myfuncs.py b/myfuncs.py index 180dfcb..8f80c3f 100644 --- a/myfuncs.py +++ b/myfuncs.py @@ -6,7 +6,7 @@ from collections.abc import Callable from functools import wraps from types import FrameType -from typing import Any, Optional, TypeVar +from typing import Any, TypeVar logging.basicConfig(format="%(asctime)s %(message)s") @@ -103,7 +103,6 @@ def daemonize() -> None: class TimedOutError(Exception): """Raised by the @timeout decorator when a function exceeds its time limit.""" - pass def timeout( @@ -126,7 +125,7 @@ def my_func(some, args): """ def decorator(func: F) -> F: - def _timed_out(signum: int, frame: Optional[FrameType]) -> None: + def _timed_out(signum: int, frame: FrameType | None) -> None: raise TimedOutError(error_message) @wraps(func) diff --git a/pong.py b/pong.py index fc4e529..516fd50 100644 --- a/pong.py +++ b/pong.py @@ -48,9 +48,7 @@ def collision(self, bally: int) -> bool: ybottom: int = int(self.ybottom) if ypos <= int(bally) <= ybottom: return True - if ypos <= ballybot <= ybottom: - return True - return False + return ypos <= ballybot <= ybottom @micropython.native def move(self, lr: bool, bally: int) -> None: @@ -178,12 +176,10 @@ def displayBall(bx: int, by: int, color: int) -> None: @micropython.native def detectCollision(bx: int, by: int, p1c: Player, p2c: Player, lr: bool) -> bool: - if bx == 20: - if p1c.collision(by): - lr = not lr - if bx == 280: - if p2c.collision(by): - lr = not lr + if bx == 20 and p1c.collision(by): + lr = not lr + if bx == 280 and p2c.collision(by): + lr = not lr return lr diff --git a/pythonrc.py b/pythonrc.py index cd860f9..94eb021 100755 --- a/pythonrc.py +++ b/pythonrc.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -# ruff: noqa: E402 - CLEAN_NS must be captured before any import + # The MIT License (MIT) # # Copyright (c) 2015-2021 Steven Fernandez @@ -23,7 +23,6 @@ # SOFTWARE. # CLEAN_NS must be captured before any imports, so imports can't be at the top. -# ruff: noqa: E402 # Keep a copy of the initial namespace, we'll need it later CLEAN_NS = globals().copy() @@ -325,10 +324,13 @@ def get_import_matches(self, text, words): names = getattr(mod, "__all__", None) or dir(mod) seen = set(matches) for name in names: - if isinstance(name, str) and name.startswith(text): - if name not in seen: - seen.add(name) - matches.append(name) + if ( + isinstance(name, str) + and name.startswith(text) + and name not in seen + ): + seen.add(name) + matches.append(name) return matches return [] @@ -445,7 +447,7 @@ def runcode_sync(self, code): return super().runcode(code) try: - exec(code, self.locals) + exec(code, self.locals) # noqa: S102 - this is a REPL except NameError as err: if self._auto_import(err, code): return self.runcode(code) diff --git a/ruff.toml b/ruff.toml new file mode 100644 index 0000000..b9f9c74 --- /dev/null +++ b/ruff.toml @@ -0,0 +1,13 @@ +target-version = "py311" + +[lint] +# This repo is a collection of standalone scripts. These rules flag +# intentional patterns used throughout them rather than bugs. +ignore = [ + "BLE001", # blind except: scripts deliberately log-and-continue + "DTZ006", # naive datetime.fromtimestamp: local time is intended + "EXE001", # shebang on non-executable file: modes are not tracked reliably + "LOG015", # root logger calls: fine for single-file scripts + "S110", # try/except/pass: used for best-effort cleanup + "UP031", # printf-style formatting: existing style +] diff --git a/summarize_git.py b/summarize_git.py index 902ea4c..7af26b5 100755 --- a/summarize_git.py +++ b/summarize_git.py @@ -10,7 +10,7 @@ import os import re import stat -from datetime import datetime, timezone +from datetime import UTC, datetime from sys import stderr from typing import Any @@ -489,7 +489,7 @@ def main() -> int: if repos: data: dict[str, Any] = { - "generated": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%SZ"), + "generated": datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%SZ"), "model": options["model"], "source_directory": os.path.abspath(basedir), "repos": repos, diff --git a/tcpQueue.py b/tcpQueue.py index 6c7d23b..d66fa52 100644 --- a/tcpQueue.py +++ b/tcpQueue.py @@ -446,7 +446,9 @@ def __init__( timeout: float = 75.0, ) -> None: if not isinstance(host, str): - raise ValueError("host must be a string, got %s" % type(host).__name__) + raise ValueError( # noqa: TRY004 - public API raises ValueError + "host must be a string, got %s" % type(host).__name__ + ) if ( isinstance(port, bool) or not isinstance(port, int) @@ -463,7 +465,7 @@ def __init__( if len(secret_key) < 16: raise ValueError("secret_key must be at least 16 bytes") if isinstance(max_queue_size, bool) or not isinstance(max_queue_size, int): - raise ValueError( + raise ValueError( # noqa: TRY004 - public API raises ValueError "max_queue_size must be an integer, got %s" % type(max_queue_size).__name__ ) diff --git a/test_heartbeat.py b/test_heartbeat.py index 6f788a5..22c03ff 100644 --- a/test_heartbeat.py +++ b/test_heartbeat.py @@ -49,13 +49,13 @@ def _wait_for(predicate, timeout=15.0, interval=0.1): def _make_pair(port_a, port_b, **kwargs): """Two heartbeats wired to each other on 127.0.0.1 with different ports. Tight intervals so tests don't take forever.""" - defaults = dict( - health_check=lambda: True, - heartbeat_interval=0.2, - heartbeat_timeout=1.0, - local_check_interval=10.0, - failover_grace=1.5, - ) + defaults = { + "health_check": lambda: True, + "heartbeat_interval": 0.2, + "heartbeat_timeout": 1.0, + "local_check_interval": 10.0, + "failover_grace": 1.5, + } defaults.update(kwargs) a = Heartbeat("127.0.0.1", "127.0.0.1", port=port_a, peer_port=port_b, **defaults) @@ -268,7 +268,7 @@ def fake_peer(): try: trailing = conn.recv(64) received.append(("trailing", trailing)) - except _sock.timeout: + except TimeoutError: received.append("timeout") conn.close() finally: @@ -524,7 +524,7 @@ def test_no_split_brain_under_simultaneous_start(): b.start() try: ok = _wait_for( - lambda: {a.get_state(), b.get_state()} == {"P", "S"}, + lambda a=a, b=b: {a.get_state(), b.get_state()} == {"P", "S"}, timeout=8, ) if not ok: diff --git a/test_quova.py b/test_quova.py index 4ffefc7..f3b4847 100644 --- a/test_quova.py +++ b/test_quova.py @@ -11,7 +11,7 @@ import threading import types import unittest -from typing import Any +from typing import Any, Self from unittest.mock import MagicMock, patch # --------------------------------------------------------------------------- @@ -38,7 +38,7 @@ def record_by_addr(self, ip: str) -> dict[str, Any] | None: class _DaemonContextStub: - def __enter__(self) -> _DaemonContextStub: + def __enter__(self) -> Self: return self def __exit__(self, *args: object) -> None: @@ -48,7 +48,7 @@ def __exit__(self, *args: object) -> None: _daemon_mod.DaemonContext = _DaemonContextStub # type: ignore[attr-defined] sys.modules.setdefault("daemon", _daemon_mod) -import quova # noqa: E402 (must come after stubs) +import quova # --------------------------------------------------------------------------- # Test fixtures @@ -144,7 +144,7 @@ def test_empty_ipinfo_uses_empty_strings(self) -> None: def test_unicode_city_name(self) -> None: info: quova.GeoIPRecord = {**SAMPLE_IPINFO, "city": "Düsseldorf"} result: bytes = quova.encode_quova(info, "1.2.3.4") - self.assertIn("düsseldorf".encode("utf-8"), result) + self.assertIn("düsseldorf".encode(), result) def test_counter_increments_per_call(self) -> None: for _ in range(5): diff --git a/vushta.py b/vushta.py index 99ea77c..fa1fc53 100644 --- a/vushta.py +++ b/vushta.py @@ -1,10 +1,9 @@ #!/usr/bin/env python3 -uO from random import randint, seed, shuffle -from typing import Optional -class ForbiddenDelight(object): +class ForbiddenDelight: """The class of all Forbidden Delights.""" def __init__(self) -> None: @@ -80,7 +79,7 @@ def __init__(self) -> None: self.things.append("a slime burger") self.things.append("a ripped magicians hat") - def Delight(self, svalue: Optional[int] = None) -> str: + def Delight(self, svalue: int | None = None) -> str: """Return a random Forbidden Delight""" seed(svalue) shuffle(self.beasts) diff --git a/web_stats.py b/web_stats.py index 3006926..1a9e157 100644 --- a/web_stats.py +++ b/web_stats.py @@ -12,13 +12,13 @@ from queue import Queue from string import ascii_letters from sys import exit as sysexit -from typing import Any, Dict, List +from typing import Any import requests from matplotlib import pyplot as plt # THe list of hosts to pull stats from -hostList: List[str] = [] +hostList: list[str] = [] # The base file path to write data base: str = "/var/tmp/web_stats" webbase: str = "/var/www/html/stats/" @@ -27,17 +27,17 @@ # Save the collected data between invocations def handler(signum, frame): logging.info("Signal received. Writing data and exiting.") - for host in host_stats: + for host, hstats in host_stats.items(): logging.info(f"Writing stats for {host}") with open(f"{base}/{host}.pickle", "wb") as f: - f.write(host_stats[host].dumps()) + f.write(hstats.dumps()) sysexit(0) # A class to make it wasy to save stats class stats: def __init__(self, hostname: str) -> None: - self.data: Dict = {} + self.data: dict = {} self.hostname: str = hostname def addType(self, valType: str) -> None: @@ -57,15 +57,15 @@ def addVal(self, valType: str, value: Any) -> None: now = int(time.time()) self.data[valType][now] = value - def retVals(self, valType: str) -> List: - retList: List = [] + def retVals(self, valType: str) -> list: + retList: list = [] if valType not in self.data: return [] for datum in self.data[valType]: retList.append([self.data[valType][datum], datum]) - return list(sorted(retList, key=lambda x: x[1])) + return sorted(retList, key=lambda x: x[1]) - def retTypes(self) -> List[str]: + def retTypes(self) -> list[str]: return list(self.data.keys()) def dumps(self) -> bytes: @@ -107,7 +107,7 @@ def pull_stats( # Take the pulled data save it and process it -def process_data(data: Dict) -> None: +def process_data(data: dict) -> None: hostname = data["hostname"] logging.debug(f"Processing data for {hostname}") myStats: stats = host_stats[hostname] @@ -125,12 +125,12 @@ def generate_graphs(hostname: str) -> None: filepath: str = f"{webbase}{hostname}_" for name in myStats.retTypes(): sstats = myStats.retVals(name) - x: List[datetime] = [] + x: list[datetime] = [] match name: case "cpu": - usercpu: List = [] - syscpu: List = [] - idlecpu: List = [] + usercpu: list = [] + syscpu: list = [] + idlecpu: list = [] x = [] for vals in sstats: x.append(datetime.fromtimestamp(vals[1])) @@ -152,9 +152,9 @@ def generate_graphs(hostname: str) -> None: plt.savefig(f"{filepath}cpu.png", format="png") plt.close() case "process_counts": - total: List[int] = [] - running: List[int] = [] - sleeping: List[int] = [] + total: list[int] = [] + running: list[int] = [] + sleeping: list[int] = [] x = [] for vals in sstats: x.append(datetime.fromtimestamp(vals[1])) @@ -171,9 +171,9 @@ def generate_graphs(hostname: str) -> None: plt.savefig(f"{filepath}processes.png", format="png") plt.close() case "load_avg": - one: List[float] = [] - five: List[float] = [] - fifteen: List[float] = [] + one: list[float] = [] + five: list[float] = [] + fifteen: list[float] = [] x = [] for vals in sstats: x.append(datetime.fromtimestamp(vals[1])) @@ -191,7 +191,7 @@ def generate_graphs(hostname: str) -> None: plt.close() case "memory": x = [] - used: List[float] = [] + used: list[float] = [] for vals in sstats: x.append(datetime.fromtimestamp(vals[1])) val = vals[0]["used"] @@ -212,9 +212,9 @@ def generate_graphs(hostname: str) -> None: case "vm": x = [] if "vsize" in sstats[0][0]: # Is this a Mac? - vsize: List[int] = [] - swapin: List[int] = [] - swapout: List[int] = [] + vsize: list[int] = [] + swapin: list[int] = [] + swapout: list[int] = [] for vals in sstats: x.append(datetime.fromtimestamp(vals[1])) swapin.append(int(vals[0]["swapin"])) @@ -230,10 +230,10 @@ def generate_graphs(hostname: str) -> None: plt.plot(x, swapin, "o-", label="Swapins") plt.plot(x, swapout, "o-", label="Swapouts") else: - vtotal: List[float] = [] - vfree: List[float] = [] - vused: List[float] = [] - avail: List[float] = [] + vtotal: list[float] = [] + vfree: list[float] = [] + vused: list[float] = [] + avail: list[float] = [] for vals in sstats: x.append(datetime.fromtimestamp(vals[1])) vtotal.append(vals[0]["total"]) @@ -268,7 +268,7 @@ def process_queue(dataQueue: Queue, qlock: threading.Lock) -> None: no_data = dataQueue.empty() qlock.release() qlock.acquire() - data: Dict = dataQueue.get() + data: dict = dataQueue.get() qlock.release() process_data(data) @@ -284,8 +284,8 @@ def process_queue(dataQueue: Queue, qlock: threading.Lock) -> None: signal.signal(signal.SIGINT, handler) signal.signal(signal.SIGTERM, handler) # Start the background threads to pull the data - host_stats: Dict[str, stats] = {} - workerList: List[threading.Thread] = [] + host_stats: dict[str, stats] = {} + workerList: list[threading.Thread] = [] dataQueue: Queue = Queue(maxsize=0) qlock: threading.Lock = threading.Lock() s: requests.Session = requests.Session()