From 070a3989a08e343e2373e2a302386db7d984dd18 Mon Sep 17 00:00:00 2001 From: David Sangrey Date: Fri, 10 Nov 2023 11:33:04 -0500 Subject: [PATCH] [2051] Prevent Typing Reversion Thanks to @norohind for catching this one! I'm not up to date on my PEP585. --- config/__init__.py | 30 ++++++------ config/darwin.py | 10 ++-- config/windows.py | 18 ++++--- plugins/coriolis.py | 9 ++-- plugins/eddn.py | 115 ++++++++++++++++++++++---------------------- plugins/edsm.py | 70 ++++++++++++++------------- plugins/edsy.py | 6 ++- plugins/inara.py | 66 ++++++++++++------------- 8 files changed, 167 insertions(+), 157 deletions(-) diff --git a/config/__init__.py b/config/__init__.py index ae08f3a0..e86a60d1 100644 --- a/config/__init__.py +++ b/config/__init__.py @@ -9,6 +9,8 @@ Windows uses the Registry to store values in a flat manner. Linux uses a file, but for commonality it's still a flat data structure. macOS uses a 'defaults' object. """ +from __future__ import annotations + __all__ = [ # defined in the order they appear in the file 'GITVERSION_FILE', @@ -42,7 +44,7 @@ import sys import traceback import warnings from abc import abstractmethod -from typing import Any, Callable, Optional, Type, TypeVar, Union, List +from typing import Any, Callable, Type, TypeVar import semantic_version from constants import GITVERSION_FILE, applongname, appname @@ -54,19 +56,19 @@ appcmdname = 'EDMC' # NB: Do *not* import this, use the functions appversion() and appversion_nobuild() _static_appversion = '5.10.0-alpha0' -_cached_version: Optional[semantic_version.Version] = None +_cached_version: semantic_version.Version | None = None copyright = '© 2015-2019 Jonathan Harris, 2020-2023 EDCD' update_feed = 'https://raw.githubusercontent.com/EDCD/EDMarketConnector/releases/edmarketconnector.xml' update_interval = 8*60*60 # Providers marked to be in debug mode. Generally this is expected to switch to sending data to a log file -debug_senders: List[str] = [] +debug_senders: list[str] = [] # TRACE logging code that should actually be used. Means not spamming it # *all* if only interested in some things. -trace_on: List[str] = [] +trace_on: list[str] = [] capi_pretend_down: bool = False -capi_debug_access_token: Optional[str] = None +capi_debug_access_token: str | None = None # This must be done here in order to avoid an import cycle with EDMCLogging. # Other code should use EDMCLogging.get_main_logger if os.getenv("EDMC_NO_UI"): @@ -232,7 +234,7 @@ class AbstractConfig(abc.ABC): self.__eddn_url = eddn_url @property - def eddn_url(self) -> Optional[str]: + def eddn_url(self) -> str | None: """ Provide the custom EDDN URL. @@ -285,9 +287,9 @@ class AbstractConfig(abc.ABC): @staticmethod def _suppress_call( - func: Callable[..., _T], exceptions: Union[Type[BaseException], List[Type[BaseException]]] = Exception, + func: Callable[..., _T], exceptions: Type[BaseException] | list[Type[BaseException]] = Exception, *args: Any, **kwargs: Any - ) -> Optional[_T]: + ) -> _T | None: if exceptions is None: exceptions = [Exception] @@ -301,8 +303,8 @@ class AbstractConfig(abc.ABC): def get( self, key: str, - default: Union[list, str, bool, int, None] = None - ) -> Union[list, str, bool, int, None]: + default: list | str | bool | int | None = None + ) -> list | str | bool | int | None: """ Return the data for the requested key, or a default. @@ -329,7 +331,7 @@ class AbstractConfig(abc.ABC): return default @abstractmethod - def get_list(self, key: str, *, default: Optional[list] = None) -> list: + def get_list(self, key: str, *, default: list | None = None) -> list: """ Return the list referred to by the given key if it exists, or the default. @@ -338,7 +340,7 @@ class AbstractConfig(abc.ABC): raise NotImplementedError @abstractmethod - def get_str(self, key: str, *, default: Optional[str] = None) -> str: + def get_str(self, key: str, *, default: str | None = None) -> str: """ Return the string referred to by the given key if it exists, or the default. @@ -351,7 +353,7 @@ class AbstractConfig(abc.ABC): raise NotImplementedError @abstractmethod - def get_bool(self, key: str, *, default: Optional[bool] = None) -> bool: + def get_bool(self, key: str, *, default: bool | None = None) -> bool: """ Return the bool referred to by the given key if it exists, or the default. @@ -391,7 +393,7 @@ class AbstractConfig(abc.ABC): raise NotImplementedError @abstractmethod - def set(self, key: str, val: Union[int, str, List[str], bool]) -> None: + def set(self, key: str, val: int | str | list[str] | bool) -> None: """ Set the given key's data to the given value. diff --git a/config/darwin.py b/config/darwin.py index 68d30306..9c15ec32 100644 --- a/config/darwin.py +++ b/config/darwin.py @@ -5,9 +5,11 @@ Copyright (c) EDCD, All Rights Reserved Licensed under the GNU General Public License. See LICENSE file. """ +from __future__ import annotations + import pathlib import sys -from typing import Any, Dict, List, Union +from typing import Any from Foundation import ( # type: ignore NSApplicationSupportDirectory, NSBundle, NSDocumentDirectory, NSSearchPathForDirectoriesInDomains, NSUserDefaults, NSUserDomainMask @@ -52,14 +54,14 @@ class MacConfig(AbstractConfig): self.default_journal_dir_path = support_path / 'Frontier Developments' / 'Elite Dangerous' self._defaults: Any = NSUserDefaults.standardUserDefaults() - self._settings: Dict[str, Union[int, str, list]] = dict( + self._settings: dict[str, int | str | list] = dict( self._defaults.persistentDomainForName_(self.identifier) or {} ) # make writeable if (out_dir := self.get_str('out_dir')) is None or not pathlib.Path(out_dir).exists(): self.set('outdir', NSSearchPathForDirectoriesInDomains(NSDocumentDirectory, NSUserDomainMask, True)[0]) - def __raw_get(self, key: str) -> Union[None, list, str, int]: + def __raw_get(self, key: str) -> None | list | str | int: """ Retrieve the raw data for the given key. @@ -143,7 +145,7 @@ class MacConfig(AbstractConfig): return res - def set(self, key: str, val: Union[int, str, List[str], bool]) -> None: + def set(self, key: str, val: int | str | list[str] | bool) -> None: """ Set the given key's data to the given value. diff --git a/config/windows.py b/config/windows.py index 94a158f1..7fb53a37 100644 --- a/config/windows.py +++ b/config/windows.py @@ -5,6 +5,8 @@ Copyright (c) EDCD, All Rights Reserved Licensed under the GNU General Public License. See LICENSE file. """ +from __future__ import annotations + import ctypes import functools import pathlib @@ -12,7 +14,7 @@ import sys import uuid import winreg from ctypes.wintypes import DWORD, HANDLE -from typing import List, Literal, Optional, Union +from typing import Literal from config import AbstractConfig, applongname, appname, logger, update_interval assert sys.platform == 'win32' @@ -32,7 +34,7 @@ CoTaskMemFree = ctypes.windll.ole32.CoTaskMemFree CoTaskMemFree.argtypes = [ctypes.c_void_p] -def known_folder_path(guid: uuid.UUID) -> Optional[str]: +def known_folder_path(guid: uuid.UUID) -> str | None: """Look up a Windows GUID to actual folder path name.""" buf = ctypes.c_wchar_p() if SHGetKnownFolderPath(ctypes.create_string_buffer(guid.bytes_le), 0, 0, ctypes.byref(buf)): @@ -117,7 +119,7 @@ class WinConfig(AbstractConfig): logger.exception('Could not open WinSparkle handle') raise - def __get_regentry(self, key: str) -> Union[None, list, str, int]: + def __get_regentry(self, key: str) -> None | list | str | int: """Access the Registry for the raw entry.""" try: value, _type = winreg.QueryValueEx(self.__reg_handle, key) @@ -138,7 +140,7 @@ class WinConfig(AbstractConfig): logger.warning(f'Registry key {key=} returned unknown type {_type=} {value=}') return None - def get_str(self, key: str, *, default: Optional[str] = None) -> str: + def get_str(self, key: str, *, default: str | None = None) -> str: """ Return the string referred to by the given key if it exists, or the default. @@ -153,7 +155,7 @@ class WinConfig(AbstractConfig): return res - def get_list(self, key: str, *, default: Optional[list] = None) -> list: + def get_list(self, key: str, *, default: list | None = None) -> list: """ Return the list referred to by the given key if it exists, or the default. @@ -183,7 +185,7 @@ class WinConfig(AbstractConfig): return res - def get_bool(self, key: str, *, default: Optional[bool] = None) -> bool: + def get_bool(self, key: str, *, default: bool | None = None) -> bool: """ Return the bool referred to by the given key if it exists, or the default. @@ -195,14 +197,14 @@ class WinConfig(AbstractConfig): return bool(res) - def set(self, key: str, val: Union[int, str, List[str], bool]) -> None: + def set(self, key: str, val: int | str | list[str] | bool) -> None: """ Set the given key's data to the given value. Implements :meth:`AbstractConfig.set`. """ # These are the types that winreg.REG_* below resolve to. - reg_type: Union[Literal[1], Literal[4], Literal[7]] + reg_type: Literal[1] | Literal[4] | Literal[7] if isinstance(val, str): reg_type = winreg.REG_SZ diff --git a/plugins/coriolis.py b/plugins/coriolis.py index 7161a4e0..283b49d8 100644 --- a/plugins/coriolis.py +++ b/plugins/coriolis.py @@ -19,6 +19,7 @@ referenced in this file (or only in any other core plugin), and if so... `build.py` TO ENSURE THE FILES ARE ACTUALLY PRESENT IN AN END-USER INSTALLATION ON WINDOWS. """ +from __future__ import annotations import base64 import gzip @@ -26,7 +27,7 @@ import io import json import tkinter as tk from tkinter import ttk -from typing import TYPE_CHECKING, Union, Optional +from typing import TYPE_CHECKING import myNotebook as nb # noqa: N813 # its not my fault. from EDMCLogging import get_main_logger from plug import show_error @@ -80,7 +81,7 @@ def plugin_start3(path: str) -> str: return 'Coriolis' -def plugin_prefs(parent: ttk.Notebook, cmdr: Optional[str], is_beta: bool) -> tk.Frame: +def plugin_prefs(parent: ttk.Notebook, cmdr: str | None, is_beta: bool) -> tk.Frame: """Set up plugin preferences.""" PADX = 10 # noqa: N806 @@ -130,7 +131,7 @@ def plugin_prefs(parent: ttk.Notebook, cmdr: Optional[str], is_beta: bool) -> tk return conf_frame -def prefs_changed(cmdr: Optional[str], is_beta: bool) -> None: +def prefs_changed(cmdr: str | None, is_beta: bool) -> None: """ Update URLs and override mode based on user preferences. @@ -175,7 +176,7 @@ def _get_target_url(is_beta: bool) -> str: return coriolis_config.normal_url -def shipyard_url(loadout, is_beta) -> Union[str, bool]: +def shipyard_url(loadout, is_beta) -> str | bool: """Return a URL for the current ship.""" # most compact representation string = json.dumps(loadout, ensure_ascii=False, sort_keys=True, separators=(',', ':')).encode('utf-8') diff --git a/plugins/eddn.py b/plugins/eddn.py index d2e7d01e..ee70d6e9 100644 --- a/plugins/eddn.py +++ b/plugins/eddn.py @@ -18,6 +18,8 @@ referenced in this file (or only in any other core plugin), and if so... `build.py` TO ENSURE THE FILES ARE ACTUALLY PRESENT IN AN END-USER INSTALLATION ON WINDOWS. """ +from __future__ import annotations + import http import itertools import json @@ -37,9 +39,6 @@ from typing import ( Iterator, Mapping, MutableMapping, - Optional, - Dict, - List, ) from typing import OrderedDict as OrderedDictT from typing import Tuple, Union @@ -86,27 +85,27 @@ class This: self.odyssey = False # Track location to add to Journal events - self.system_address: Optional[str] = None - self.system_name: Optional[str] = None - self.coordinates: Optional[Tuple] = None - self.body_name: Optional[str] = None - self.body_id: Optional[int] = None - self.body_type: Optional[int] = None - self.station_name: Optional[str] = None - self.station_type: Optional[str] = None - self.station_marketid: Optional[str] = None + self.system_address: str | None = None + self.system_name: str | None = None + self.coordinates: tuple | None = None + self.body_name: str | None = None + self.body_id: int | None = None + self.body_type: int | None = None + self.station_name: str | None = None + self.station_type: str | None = None + self.station_marketid: str | None = None # Track Status.json data - self.status_body_name: Optional[str] = None + self.status_body_name: str | None = None # Avoid duplicates - self.marketId: Optional[str] = None - self.commodities: Optional[List[OrderedDictT[str, Any]]] = None - self.outfitting: Optional[Tuple[bool, List[str]]] = None - self.shipyard: Optional[Tuple[bool, List[Mapping[str, Any]]]] = None + self.marketId: str | None = None + self.commodities: list[OrderedDictT[str, Any]] | None = None + self.outfitting: Tuple[bool, list[str]] | None = None + self.shipyard: Tuple[bool, list[Mapping[str, Any]]] | None = None self.fcmaterials_marketid: int = 0 - self.fcmaterials: Optional[List[OrderedDictT[str, Any]]] = None + self.fcmaterials: list[OrderedDictT[str, Any]] | None = None self.fcmaterials_capi_marketid: int = 0 - self.fcmaterials_capi: Optional[List[OrderedDictT[str, Any]]] = None + self.fcmaterials_capi: list[OrderedDictT[str, Any]] | None = None # For the tkinter parent window, so we can call update_idletasks() self.parent: tk.Tk @@ -395,7 +394,7 @@ class EDDNSender: """ logger.trace_if("plugin.eddn.send", "Sending message") should_return: bool - new_data: Dict[str, Any] + new_data: dict[str, Any] should_return, new_data = killswitch.check_killswitch('plugins.eddn.send', json.loads(msg)) if should_return: @@ -404,7 +403,7 @@ class EDDNSender: # Even the smallest possible message compresses somewhat, so always compress encoded, compressed = text.gzip(json.dumps(new_data, separators=(',', ':')), max_size=0) - headers: Optional[Dict[str, str]] = None + headers: dict[str, str] | None = None if compressed: headers = {'Content-Encoding': 'gzip'} @@ -612,7 +611,7 @@ class EDDN: self.sender = EDDNSender(self, self.eddn_url) - self.fss_signals: List[Mapping[str, Any]] = [] + self.fss_signals: list[Mapping[str, Any]] = [] def close(self): """Close down the EDDN class instance.""" @@ -636,7 +635,7 @@ class EDDN: :param is_beta: whether or not we're currently in beta mode """ should_return: bool - new_data: Dict[str, Any] + new_data: dict[str, Any] should_return, new_data = killswitch.check_killswitch('capi.request./market', {}) if should_return: logger.warning("capi.request./market has been disabled by killswitch. Returning.") @@ -653,7 +652,7 @@ class EDDN: modules, ships ) - commodities: List[OrderedDictT[str, Any]] = [] + commodities: list[OrderedDictT[str, Any]] = [] for commodity in data['lastStarport'].get('commodities') or []: # Check 'marketable' and 'not prohibited' if (category_map.get(commodity['categoryname'], True) @@ -726,7 +725,7 @@ class EDDN: :param data: The raw CAPI data. :return: Sanity-checked data. """ - modules: Dict[str, Any] = data['lastStarport'].get('modules') + modules: dict[str, Any] = data['lastStarport'].get('modules') if modules is None or not isinstance(modules, dict): if modules is None: logger.debug('modules was None. FC or Damaged Station?') @@ -743,13 +742,13 @@ class EDDN: # Set a safe value modules = {} - ships: Dict[str, Any] = data['lastStarport'].get('ships') + ships: dict[str, Any] = data['lastStarport'].get('ships') if ships is None or not isinstance(ships, dict): if ships is None: logger.debug('ships was None') else: - logger.error(f'ships was neither None nor a Dict! Type = {type(ships)}') + logger.error(f'ships was neither None nor a dict! Type = {type(ships)}') # Set a safe value ships = {'shipyard_list': {}, 'unavailable_list': []} @@ -769,7 +768,7 @@ class EDDN: :param is_beta: whether or not we're currently in beta mode """ should_return: bool - new_data: Dict[str, Any] + new_data: dict[str, Any] should_return, new_data = killswitch.check_killswitch('capi.request./shipyard', {}) if should_return: logger.warning("capi.request./shipyard has been disabled by killswitch. Returning.") @@ -796,7 +795,7 @@ class EDDN: modules.values() ) - outfitting: List[str] = sorted( + outfitting: list[str] = sorted( self.MODULE_RE.sub(lambda match: match.group(0).capitalize(), mod['name'].lower()) for mod in to_search ) @@ -837,7 +836,7 @@ class EDDN: :param is_beta: whether or not we are in beta mode """ should_return: bool - new_data: Dict[str, Any] + new_data: dict[str, Any] should_return, new_data = killswitch.check_killswitch('capi.request./shipyard', {}) if should_return: logger.warning("capi.request./shipyard has been disabled by killswitch. Returning.") @@ -856,7 +855,7 @@ class EDDN: ships ) - shipyard: List[Mapping[str, Any]] = sorted( + shipyard: list[Mapping[str, Any]] = sorted( itertools.chain( (ship['name'].lower() for ship in (ships['shipyard_list'] or {}).values()), (ship['name'].lower() for ship in ships['unavailable_list'] or {}), @@ -899,8 +898,8 @@ class EDDN: :param is_beta: whether or not we're in beta mode :param entry: the journal entry containing the commodities data """ - items: List[Mapping[str, Any]] = entry.get('Items') or [] - commodities: List[OrderedDictT[str, Any]] = sorted((OrderedDict([ + items: list[Mapping[str, Any]] = entry.get('Items') or [] + commodities: list[OrderedDictT[str, Any]] = sorted((OrderedDict([ ('name', self.canonicalise(commodity['Name'])), ('meanPrice', commodity['MeanPrice']), ('buyPrice', commodity['BuyPrice']), @@ -947,11 +946,11 @@ class EDDN: :param is_beta: Whether or not we're in beta mode :param entry: The relevant journal entry """ - modules: List[Mapping[str, Any]] = entry.get('Items', []) + modules: list[Mapping[str, Any]] = entry.get('Items', []) horizons: bool = entry.get('Horizons', False) # outfitting = sorted([self.MODULE_RE.sub(lambda m: m.group(0).capitalize(), module['Name']) # for module in modules if module['Name'] != 'int_planetapproachsuite']) - outfitting: List[str] = sorted( + outfitting: list[str] = sorted( self.MODULE_RE.sub(lambda m: m.group(0).capitalize(), mod['Name']) for mod in filter(lambda m: m['Name'] != 'int_planetapproachsuite', modules) ) @@ -986,7 +985,7 @@ class EDDN: :param is_beta: Whether or not we're in beta mode :param entry: the relevant journal entry """ - ships: List[Mapping[str, Any]] = entry.get('PriceList') or [] + ships: list[Mapping[str, Any]] = entry.get('Pricelist') or [] horizons: bool = entry.get('Horizons', False) shipyard = sorted(ship['ShipType'] for ship in ships) # Don't send empty ships list - shipyard data is only guaranteed present if user has visited the shipyard. @@ -1042,7 +1041,7 @@ class EDDN: self.sender.send_message_by_id(msg_id) def standard_header( - self, game_version: Optional[str] = None, game_build: Optional[str] = None + self, game_version: str | None = None, game_build: str | None = None ) -> MutableMapping[str, Any]: """ Return the standard header for an EDDN message, given tracked state. @@ -1134,7 +1133,7 @@ class EDDN: def export_journal_fssdiscoveryscan( self, cmdr: str, system_name: str, system_starpos: list, is_beta: bool, entry: Mapping[str, Any] - ) -> Optional[str]: + ) -> str | None: """ Send an FSSDiscoveryScan to EDDN on the correct schema. @@ -1176,7 +1175,7 @@ class EDDN: def export_journal_navbeaconscan( self, cmdr: str, system_name: str, system_starpos: list, is_beta: bool, entry: Mapping[str, Any] - ) -> Optional[str]: + ) -> str | None: """ Send an NavBeaconScan to EDDN on the correct schema. @@ -1218,7 +1217,7 @@ class EDDN: def export_journal_codexentry( # noqa: CCR001 self, cmdr: str, system_starpos: list, is_beta: bool, entry: MutableMapping[str, Any] - ) -> Optional[str]: + ) -> str | None: """ Send a CodexEntry to EDDN on the correct schema. @@ -1320,7 +1319,7 @@ class EDDN: def export_journal_scanbarycentre( self, cmdr: str, system_starpos: list, is_beta: bool, entry: Mapping[str, Any] - ) -> Optional[str]: + ) -> str | None: """ Send a ScanBaryCentre to EDDN on the correct schema. @@ -1374,7 +1373,7 @@ class EDDN: def export_journal_navroute( self, cmdr: str, is_beta: bool, entry: MutableMapping[str, Any] - ) -> Optional[str]: + ) -> str | None: """ Send a NavRoute to EDDN on the correct schema. @@ -1447,7 +1446,7 @@ class EDDN: def export_journal_fcmaterials( self, cmdr: str, is_beta: bool, entry: MutableMapping[str, Any] - ) -> Optional[str]: + ) -> str | None: """ Send an FCMaterials message to EDDN on the correct schema. @@ -1531,7 +1530,7 @@ class EDDN: def export_capi_fcmaterials( self, data: CAPIData, is_beta: bool, horizons: bool - ) -> Optional[str]: + ) -> str | None: """ Send CAPI-sourced 'onfootmicroresources' data on `fcmaterials/1` schema. @@ -1594,7 +1593,7 @@ class EDDN: def export_journal_approachsettlement( self, cmdr: str, system_name: str, system_starpos: list, is_beta: bool, entry: MutableMapping[str, Any] - ) -> Optional[str]: + ) -> str | None: """ Send an ApproachSettlement to EDDN on the correct schema. @@ -1669,7 +1668,7 @@ class EDDN: def export_journal_fssallbodiesfound( self, cmdr: str, system_name: str, system_starpos: list, is_beta: bool, entry: MutableMapping[str, Any] - ) -> Optional[str]: + ) -> str | None: """ Send an FSSAllBodiesFound message to EDDN on the correct schema. @@ -1719,7 +1718,7 @@ class EDDN: def export_journal_fssbodysignals( self, cmdr: str, system_name: str, system_starpos: list, is_beta: bool, entry: MutableMapping[str, Any] - ) -> Optional[str]: + ) -> str | None: """ Send an FSSBodySignals message to EDDN on the correct schema. @@ -1789,7 +1788,7 @@ class EDDN: def export_journal_fsssignaldiscovered( self, cmdr: str, system_name: str, system_starpos: list, is_beta: bool, entry: MutableMapping[str, Any] - ) -> Optional[str]: + ) -> str | None: """ Send an FSSSignalDiscovered message to EDDN on the correct schema. @@ -1892,7 +1891,7 @@ class EDDN: match = self.CANONICALISE_RE.match(item) return match and match.group(1) or item - def capi_gameversion_from_host_endpoint(self, capi_host: Optional[str], capi_endpoint: str) -> str: + def capi_gameversion_from_host_endpoint(self, capi_host: str | None, capi_endpoint: str) -> str: """ Return the correct CAPI gameversion string for the given host/endpoint. @@ -1910,7 +1909,7 @@ class EDDN: gv = 'CAPI-Legacy-' else: - # Technically incorrect, but it will inform Listeners + # Technically incorrect, but it will inform listeners logger.error(f"{capi_host=} lead to bad gameversion") gv = 'CAPI-UNKNOWN-' ####################################################################### @@ -1924,7 +1923,7 @@ class EDDN: gv += 'shipyard' else: - # Technically incorrect, but it will inform Listeners + # Technically incorrect, but it will inform listeners logger.error(f"{capi_endpoint=} lead to bad gameversion") gv += 'UNKNOWN' ####################################################################### @@ -1943,7 +1942,7 @@ def plugin_start3(plugin_dir: str) -> str: return 'EDDN' -def plugin_app(parent: tk.Tk) -> Optional[tk.Frame]: +def plugin_app(parent: tk.Tk) -> tk.Frame | None: """ Set up any plugin-specific UI. @@ -2183,7 +2182,7 @@ def filter_localised(d: Mapping[str, Any]) -> OrderedDictT[str, Any]: """ Recursively remove any dict keys with names ending `_Localised` from a dict. - :param d: Dict to filter keys of. + :param d: dict to filter keys of. :return: The filtered dict. """ filtered: OrderedDictT[str, Any] = OrderedDict() @@ -2207,7 +2206,7 @@ def capi_filter_localised(d: Mapping[str, Any]) -> OrderedDictT[str, Any]: """ Recursively remove any dict keys for known CAPI 'localised' names. - :param d: Dict to filter keys of. + :param d: dict to filter keys of. :return: The filtered dict. """ filtered: OrderedDictT[str, Any] = OrderedDict() @@ -2234,7 +2233,7 @@ def journal_entry( # noqa: C901, CCR001 station: str, entry: MutableMapping[str, Any], state: Mapping[str, Any] -) -> Optional[str]: +) -> str | None: """ Process a new Journal entry. @@ -2491,7 +2490,7 @@ def journal_entry( # noqa: C901, CCR001 return None -def cmdr_data_legacy(data: CAPIData, is_beta: bool) -> Optional[str]: +def cmdr_data_legacy(data: CAPIData, is_beta: bool) -> str | None: """ Process new CAPI data for Legacy galaxy. @@ -2510,7 +2509,7 @@ def cmdr_data_legacy(data: CAPIData, is_beta: bool) -> Optional[str]: return cmdr_data(data, is_beta) -def cmdr_data(data: CAPIData, is_beta: bool) -> Optional[str]: # noqa: CCR001 +def cmdr_data(data: CAPIData, is_beta: bool) -> str | None: # noqa: CCR001 """ Process new CAPI data for not-Legacy galaxy (might be beta). @@ -2611,7 +2610,7 @@ def capi_is_horizons(economies: MAP_STR_ANY, modules: MAP_STR_ANY, ships: MAP_ST return economies_colony or modules_horizons or ship_horizons -def dashboard_entry(cmdr: str, is_beta: bool, entry: Dict[str, Any]) -> None: +def dashboard_entry(cmdr: str, is_beta: bool, entry: dict[str, Any]) -> None: """ Process Status.json data to track things like current Body. diff --git a/plugins/edsm.py b/plugins/edsm.py index 2f644dcb..33af692b 100644 --- a/plugins/edsm.py +++ b/plugins/edsm.py @@ -18,6 +18,8 @@ referenced in this file (or only in any other core plugin), and if so... `build.py` TO ENSURE THE FILES ARE ACTUALLY PRESENT IN AN END-USER INSTALLATION ON WINDOWS. """ +from __future__ import annotations + import json import threading import tkinter as tk @@ -26,7 +28,7 @@ from queue import Queue from threading import Thread from time import sleep from tkinter import ttk -from typing import TYPE_CHECKING, Any, Dict, List, Literal, Mapping, MutableMapping, Optional, Set, Tuple, Union, cast +from typing import TYPE_CHECKING, Any, Literal, Mapping, MutableMapping, cast import requests import killswitch import monitor @@ -72,27 +74,27 @@ class This: self.game_build = "" # Handle only sending Live galaxy data - self.legacy_galaxy_last_notified: Optional[datetime] = None + self.legacy_galaxy_last_notified: datetime | None = None self.session: requests.Session = requests.Session() self.session.headers['User-Agent'] = user_agent self.queue: Queue = Queue() # Items to be sent to EDSM by worker thread - self.discarded_events: Set[str] = set() # List discarded events from EDSM - self.lastlookup: Dict[str, Any] # Result of last system lookup + self.discarded_events: set[str] = set() # List discarded events from EDSM + self.lastlookup: dict[str, Any] # Result of last system lookup # Game state self.multicrew: bool = False # don't send captain's ship info to EDSM while on a crew - self.coordinates: Optional[Tuple[int, int, int]] = None + self.coordinates: tuple[int, int, int] | None = None self.newgame: bool = False # starting up - batch initial burst of events self.newgame_docked: bool = False # starting up while docked self.navbeaconscan: int = 0 # batch up burst of Scan events after NavBeaconScan - self.system_link: Optional[tk.Widget] = None - self.system_name: Optional[tk.Tk] = None - self.system_address: Optional[int] = None # Frontier SystemAddress - self.system_population: Optional[int] = None - self.station_link: Optional[tk.Widget] = None - self.station_name: Optional[str] = None - self.station_marketid: Optional[int] = None # Frontier MarketID + self.system_link: tk.Widget | None = None + self.system_name: tk.Tk | None = None + self.system_address: int | None = None # Frontier SystemAddress + self.system_population: int | None = None + self.station_link: tk.Widget | None = None + self.station_name: str | None = None + self.station_marketid: int | None = None # Frontier MarketID self.on_foot = False self._IMG_KNOWN = None @@ -100,21 +102,21 @@ class This: self._IMG_NEW = None self._IMG_ERROR = None - self.thread: Optional[threading.Thread] = None + self.thread: threading.Thread | None = None - self.log: Optional[tk.IntVar] = None - self.log_button: Optional[ttk.Checkbutton] = None + self.log: tk.IntVar | None = None + self.log_button: ttk.Checkbutton | None = None - self.label: Optional[tk.Widget] = None + self.label: tk.Widget | None = None - self.cmdr_label: Optional[nb.Label] = None - self.cmdr_text: Optional[nb.Label] = None + self.cmdr_label: nb.Label | None = None + self.cmdr_text: nb.Label | None = None - self.user_label: Optional[nb.Label] = None - self.user: Optional[nb.Entry] = None + self.user_label: nb.Label | None = None + self.user: nb.Entry | None = None - self.apikey_label: Optional[nb.Label] = None - self.apikey: Optional[nb.Entry] = None + self.apikey_label: nb.Label | None = None + self.apikey: nb.Entry | None = None this = This() @@ -277,7 +279,7 @@ def toggle_password_visibility(): this.apikey.config(show="*") # type: ignore -def plugin_prefs(parent: ttk.Notebook, cmdr: Optional[str], is_beta: bool) -> tk.Frame: +def plugin_prefs(parent: ttk.Notebook, cmdr: str | None, is_beta: bool) -> tk.Frame: """ Plugin preferences setup hook. @@ -361,7 +363,7 @@ def plugin_prefs(parent: ttk.Notebook, cmdr: Optional[str], is_beta: bool) -> tk return frame -def prefs_cmdr_changed(cmdr: Optional[str], is_beta: bool) -> None: # noqa: CCR001 +def prefs_cmdr_changed(cmdr: str | None, is_beta: bool) -> None: # noqa: CCR001 """ Handle the Commander name changing whilst Settings was open. @@ -390,7 +392,7 @@ def prefs_cmdr_changed(cmdr: Optional[str], is_beta: bool) -> None: # noqa: CCR # LANG: We have no data on the current commander this.cmdr_text['text'] = _('None') - to_set: Union[Literal['normal'], Literal['disabled']] = tk.DISABLED + to_set: Literal['normal'] | Literal['disabled'] = tk.DISABLED if cmdr and not is_beta and this.log and this.log.get(): to_set = tk.NORMAL @@ -440,9 +442,9 @@ def prefs_changed(cmdr: str, is_beta: bool) -> None: config.set('edsm_out', this.log.get()) if cmdr and not is_beta: - cmdrs: List[str] = config.get_list('edsm_cmdrs', default=[]) - usernames: List[str] = config.get_list('edsm_usernames', default=[]) - apikeys: List[str] = config.get_list('edsm_apikeys', default=[]) + cmdrs: list[str] = config.get_list('edsm_cmdrs', default=[]) + usernames: list[str] = config.get_list('edsm_usernames', default=[]) + apikeys: list[str] = config.get_list('edsm_apikeys', default=[]) if this.user and this.apikey: if cmdr in cmdrs: @@ -460,7 +462,7 @@ def prefs_changed(cmdr: str, is_beta: bool) -> None: config.set('edsm_apikeys', apikeys) -def credentials(cmdr: str) -> Optional[Tuple[str, str]]: +def credentials(cmdr: str) -> tuple[str, str] | None: """ Get credentials for the given commander, if they exist. @@ -635,7 +637,7 @@ Queueing: {entry!r}''' # Update system data -def cmdr_data(data: CAPIData, is_beta: bool) -> Optional[str]: # noqa: CCR001 +def cmdr_data(data: CAPIData, is_beta: bool) -> str | None: # noqa: CCR001 """ Process new CAPI data. @@ -722,7 +724,7 @@ def worker() -> None: # noqa: CCR001 C901 :return: None """ logger.debug('Starting...') - pending: List[Mapping[str, Any]] = [] # Unsent events + pending: list[Mapping[str, Any]] = [] # Unsent events closing = False cmdr: str = "" last_game_version = "" @@ -744,7 +746,7 @@ def worker() -> None: # noqa: CCR001 C901 logger.debug(f'{this.shutting_down=}, so setting closing = True') closing = True - item: Optional[Tuple[str, str, str, Mapping[str, Any]]] = this.queue.get() + item: tuple[str, str, str, Mapping[str, Any]] | None = this.queue.get() if item: (cmdr, game_version, game_build, entry) = item logger.trace_if(CMDR_EVENTS, f'De-queued ({cmdr=}, {game_version=}, {game_build=}, {entry["event"]=})') @@ -756,7 +758,7 @@ def worker() -> None: # noqa: CCR001 C901 retrying = 0 while retrying < 3: if item is None: - item = cast(Tuple[str, str, str, Mapping[str, Any]], ("", {})) + item = cast(tuple[str, str, str, Mapping[str, Any]], ("", {})) should_skip, new_item = killswitch.check_killswitch( 'plugins.edsm.worker', item, @@ -909,7 +911,7 @@ def worker() -> None: # noqa: CCR001 C901 last_game_build = game_build -def should_send(entries: List[Mapping[str, Any]], event: str) -> bool: # noqa: CCR001 +def should_send(entries: list[Mapping[str, Any]], event: str) -> bool: # noqa: CCR001 """ Whether or not any of the given entries should be sent to EDSM. diff --git a/plugins/edsy.py b/plugins/edsy.py index 0c78a429..a02d3424 100644 --- a/plugins/edsy.py +++ b/plugins/edsy.py @@ -18,11 +18,13 @@ referenced in this file (or only in any other core plugin), and if so... `build.py` TO ENSURE THE FILES ARE ACTUALLY PRESENT IN AN END-USER INSTALLATION ON WINDOWS. """ +from __future__ import annotations + import base64 import gzip import io import json -from typing import Any, Mapping, Union +from typing import Any, Mapping def plugin_start3(plugin_dir: str) -> str: @@ -36,7 +38,7 @@ def plugin_start3(plugin_dir: str) -> str: # Return a URL for the current ship -def shipyard_url(loadout: Mapping[str, Any], is_beta: bool) -> Union[bool, str]: +def shipyard_url(loadout: Mapping[str, Any], is_beta: bool) -> bool | str: """ Construct a URL for ship loadout. diff --git a/plugins/inara.py b/plugins/inara.py index efe010df..2a124f54 100644 --- a/plugins/inara.py +++ b/plugins/inara.py @@ -18,6 +18,7 @@ referenced in this file (or only in any other core plugin), and if so... `build.py` TO ENSURE THE FILES ARE ACTUALLY PRESENT IN AN END-USER INSTALLATION ON WINDOWS. """ +from __future__ import annotations import json import threading @@ -29,9 +30,8 @@ from datetime import datetime, timedelta, timezone from operator import itemgetter from threading import Lock, Thread from tkinter import ttk -from typing import TYPE_CHECKING, Any, Callable, Deque, Dict, List, Mapping, NamedTuple, Optional +from typing import TYPE_CHECKING, Any, Callable, Deque, Mapping, NamedTuple, Sequence, cast, Union from typing import OrderedDict as OrderedDictT -from typing import Sequence, Union, cast import requests import edmc_data import killswitch @@ -63,8 +63,8 @@ CREDITS_DELTA_MIN_ABSOLUTE = 10_000_000 # Absolute difference threshold class Credentials(NamedTuple): """Credentials holds the set of credentials required to identify an inara API payload to inara.""" - cmdr: Optional[str] - fid: Optional[str] + cmdr: str | None + fid: str | None api_key: str @@ -89,25 +89,25 @@ class This: self.parent: tk.Tk # Handle only sending Live galaxy data - self.legacy_galaxy_last_notified: Optional[datetime] = None + self.legacy_galaxy_last_notified: datetime | None = None self.lastlocation = None # eventData from the last Commander's Flight Log event self.lastship = None # eventData from the last addCommanderShip or setCommanderShip event # Cached Cmdr state - self.cmdr: Optional[str] = None - self.FID: Optional[str] = None # Frontier ID + self.cmdr: str | None = None + self.FID: str | None = None # Frontier ID self.multicrew: bool = False # don't send captain's ship info to Inara while on a crew self.newuser: bool = False # just entered API Key - send state immediately self.newsession: bool = True # starting a new session - wait for Cargo event self.undocked: bool = False # just undocked self.suppress_docked = False # Skip initial Docked event if started docked - self.cargo: Optional[List[OrderedDictT[str, Any]]] = None - self.materials: Optional[List[OrderedDictT[str, Any]]] = None + self.cargo: list[OrderedDictT[str, Any]] | None = None + self.materials: list[OrderedDictT[str, Any]] | None = None self.last_credits: int = 0 # Send credit update soon after Startup / new game - self.storedmodules: Optional[List[OrderedDictT[str, Any]]] = None - self.loadout: Optional[OrderedDictT[str, Any]] = None - self.fleet: Optional[List[OrderedDictT[str, Any]]] = None + self.storedmodules: list[OrderedDictT[str, Any]] | None = None + self.loadout: OrderedDictT[str, Any] | None = None + self.fleet: list[OrderedDictT[str, Any]] | None = None self.shipswap: bool = False # just swapped ship self.on_foot = False @@ -115,9 +115,9 @@ class This: # Main window clicks self.system_link: tk.Widget = None # type: ignore - self.system_name: Optional[str] = None # type: ignore - self.system_address: Optional[str] = None # type: ignore - self.system_population: Optional[int] = None + self.system_name: str | None = None # type: ignore + self.system_address: str | None = None # type: ignore + self.system_population: int | None = None self.station_link: tk.Widget = None # type: ignore self.station = None self.station_marketid = None @@ -129,7 +129,7 @@ class This: self.apikey: nb.Entry self.apikey_label: tk.Label - self.events: Dict[Credentials, Deque[Event]] = defaultdict(deque) + self.events: dict[Credentials, Deque[Event]] = defaultdict(deque) self.event_lock: Lock = threading.Lock() # protects events, for use when rewriting events def filter_events(self, key: Credentials, predicate: Callable[[Event], bool]) -> None: @@ -361,7 +361,7 @@ def prefs_changed(cmdr: str, is_beta: bool) -> None: ) -def credentials(cmdr: Optional[str]) -> Optional[str]: +def credentials(cmdr: str | None) -> str | None: """ Get the credentials for the current commander. @@ -383,7 +383,7 @@ def credentials(cmdr: Optional[str]) -> Optional[str]: def journal_entry( # noqa: C901, CCR001 - cmdr: str, is_beta: bool, system: str, station: str, entry: Dict[str, Any], state: Dict[str, Any] + cmdr: str, is_beta: bool, system: str, station: str, entry: dict[str, Any], state: dict[str, Any] ) -> str: """ Journal entry hook. @@ -394,7 +394,7 @@ def journal_entry( # noqa: C901, CCR001 # causing users to spam Inara with 'URL provider' queries, and we want to # stop that. should_return: bool - new_entry: Dict[str, Any] = {} + new_entry: dict[str, Any] = {} should_return, new_entry = killswitch.check_killswitch('plugins.inara.journal', entry, logger) if should_return: @@ -813,7 +813,7 @@ def journal_entry( # noqa: C901, CCR001 # Fleet if event_name == 'StoredShips': - fleet: List[OrderedDictT[str, Any]] = sorted( + fleet: list[OrderedDictT[str, Any]] = sorted( [OrderedDict({ 'shipType': x['ShipType'], 'shipGameID': x['ShipID'], @@ -860,7 +860,7 @@ def journal_entry( # noqa: C901, CCR001 # Stored modules if event_name == 'StoredModules': items = {mod['StorageSlot']: mod for mod in entry['Items']} # Impose an order - modules: List[OrderedDictT[str, Any]] = [] + modules: list[OrderedDictT[str, Any]] = [] for slot in sorted(items): item = items[slot] module: OrderedDictT[str, Any] = OrderedDict([ @@ -1088,7 +1088,7 @@ def journal_entry( # noqa: C901, CCR001 # # So we're going to do a lot of checking here and bail out if we dont like the look of ANYTHING here - to_send_data: Optional[Dict[str, Any]] = {} # This is a glorified sentinel until lower down. + to_send_data: dict[str, Any] | None = {} # This is a glorified sentinel until lower down. # On Horizons, neither of these exist on TouchDown star_system_name = entry.get('StarSystem', this.system_name) body_name = entry.get('Body', state['Body'] if state['BodyType'] == 'Planet' else None) @@ -1370,7 +1370,7 @@ def cmdr_data(data: CAPIData, is_beta): # noqa: CCR001, reanalyze me later pass -def make_loadout(state: Dict[str, Any]) -> OrderedDictT[str, Any]: # noqa: CCR001 +def make_loadout(state: dict[str, Any]) -> OrderedDictT[str, Any]: # noqa: CCR001 """ Construct an inara loadout from an event. @@ -1440,8 +1440,8 @@ def new_add_event( name: str, timestamp: str, data: EVENT_DATA, - cmdr: Optional[str] = None, - fid: Optional[str] = None + cmdr: str | None = None, + fid: str | None = None ): """ Add a journal event to the queue, to be sent to inara at the next opportunity. @@ -1470,11 +1470,11 @@ def new_add_event( this.events[key].append(Event(name, timestamp, data)) -def clean_event_list(event_list: List[Event]) -> List[Event]: +def clean_event_list(event_list: list[Event]) -> list[Event]: """ Check for killswitched events and remove or modify them as requested. - :param event_list: List of events to clean + :param event_list: list of events to clean :return: Cleaned list of events """ cleaned_events = [] @@ -1533,14 +1533,14 @@ def new_worker(): logger.debug('Done.') -def get_events(clear: bool = True) -> Dict[Credentials, List[Event]]: +def get_events(clear: bool = True) -> dict[Credentials, list[Event]]: """ Fetch a copy of all events from the current queue. :param clear: whether to clear the queues as we go, defaults to True :return: a copy of the event dictionary """ - events_copy: Dict[Credentials, List[Event]] = {} + events_copy: dict[Credentials, list[Event]] = {} with this.event_lock: for key, events in this.events.items(): @@ -1590,7 +1590,7 @@ def send_data(url: str, data: Mapping[str, Any]) -> bool: return True # Regardless of errors above, we DID manage to send it, therefore inform our caller as such -def handle_api_error(data: Mapping[str, Any], status: int, reply: Dict[str, Any]) -> None: +def handle_api_error(data: Mapping[str, Any], status: int, reply: dict[str, Any]) -> None: """ Handle API error response. @@ -1604,7 +1604,7 @@ def handle_api_error(data: Mapping[str, Any], status: int, reply: Dict[str, Any] plug.show_error(_('Error: Inara {MSG}').format(MSG=error_message)) -def handle_success_reply(data: Mapping[str, Any], reply: Dict[str, Any]) -> None: +def handle_success_reply(data: Mapping[str, Any], reply: dict[str, Any]) -> None: """ Handle successful API response. @@ -1619,7 +1619,7 @@ def handle_success_reply(data: Mapping[str, Any], reply: Dict[str, Any]) -> None handle_special_events(data_event, reply_event) -def handle_individual_error(data_event: Dict[str, Any], reply_status: int, reply_text: str) -> None: +def handle_individual_error(data_event: dict[str, Any], reply_status: int, reply_text: str) -> None: """ Handle individual API error. @@ -1638,7 +1638,7 @@ def handle_individual_error(data_event: Dict[str, Any], reply_status: int, reply )) -def handle_special_events(data_event: Dict[str, Any], reply_event: Dict[str, Any]) -> None: +def handle_special_events(data_event: dict[str, Any], reply_event: dict[str, Any]) -> None: """ Handle special events in the API response.