diff --git a/coriolis-update-files.py b/coriolis-update-files.py index f60eff75..acddf816 100755 --- a/coriolis-update-files.py +++ b/coriolis-update-files.py @@ -1,6 +1,5 @@ #!/usr/bin/env python3 -""" -coriolis-update-files.py - Build ship and module databases from https://github.com/EDCD/coriolis-data/. +"""coriolis-update-files.py - Build ship and module databases from coriolis-data. Copyright (c) EDCD, All Rights Reserved Licensed under the GNU General Public License v2 or later. @@ -12,86 +11,101 @@ project structure is used for this purpose. If you want to utilize the FDevIDs/ version of the file, copy it over the local one. """ - import json import subprocess import sys +from typing import Any import outfitting from edmc_data import coriolis_ship_map, ship_name_map -if __name__ == "__main__": # noqa: C901 - def add(modules, name, attributes) -> None: - """Add the given module to the modules dict.""" - if name in modules and modules[name] != attributes: - raise ValueError(f'{name}: {modules.get(name)} != {attributes}') - if name in modules: - raise ValueError(f'{name} already exists in modules') +def add_module(modules: dict[str, Any], name: str, attributes: dict[str, Any]) -> None: + """Add the given module to the modules dict safely.""" + if name in modules: + if modules[name] != attributes: + raise ValueError(f"{name}: {modules[name]} != {attributes}") + raise ValueError(f"{name} already exists in modules") - modules[name] = attributes + modules[name] = attributes + +def main() -> None: # noqa: CCR001 + """Run the Coriolis Updater.""" try: # Regenerate coriolis-data distribution - subprocess.check_call('npm install', cwd='coriolis-data', shell=True, stdout=sys.stdout, stderr=sys.stderr) + subprocess.run( + "npm install", cwd="coriolis-data", shell=True, check=True, stdout=sys.stdout, stderr=sys.stderr, + ) except NotADirectoryError: - sys.exit("Coriolis-Data Directory not found! Have you set up your submodules? \n" - "https://github.com/EDCD/EDMarketConnector/wiki/Running-from-source#obtain-a-copy-of-the-application-source") # noqa: E501 + sys.exit( + "Coriolis-Data Directory not found! Have you set up your submodules? \n" + "https://github.com/EDCD/EDMarketConnector/wiki/Running-from-source#obtain-a-copy-of-the-application-source" + ) - file_path = 'coriolis-data/dist/index.json' - with open(file_path) as file: + file_path = "coriolis-data/dist/index.json" + with open(file_path, encoding="utf-8") as file: data = json.load(file) # Symbolic name from in-game name - reverse_ship_map = {v: k for k, v in list(ship_name_map.items())} - + reverse_ship_map = {v: k for k, v in ship_name_map.items()} bulkheads = list(outfitting.armour_map) - ships = {} - modules = {} + ships: dict[str, Any] = {} + modules: dict[str, Any] = {} - # Ship and armour masses - for m in list(data['Ships'].values()): - name = coriolis_ship_map.get(m['properties']['name'], str(m['properties']['name'])) + # Process Ship and armour masses + for m in data["Ships"].values(): + raw_name = m["properties"]["name"] + name = coriolis_ship_map.get(raw_name, str(raw_name)) if name not in reverse_ship_map: - raise ValueError(f'Unknown ship: {name}') - ships[name] = {'hullMass': m['properties']['hullMass'], - 'reserveFuelCapacity': m['properties']['reserveFuelCapacity']} - for i, bulkhead in enumerate(bulkheads): - modules['_'.join([reverse_ship_map[name], 'armour', bulkhead])] = {'mass': m['bulkheads'][i]['mass']} + raise ValueError(f"Unknown ship: {name}") - ships = {k: ships[k] for k in sorted(ships)} - with open("ships.json", "w") as ships_file: + ships[name] = {"hullMass": m["properties"]["hullMass"], + "reserveFuelCapacity": m["properties"]["reserveFuelCapacity"]} + + for i, bulkhead in enumerate(bulkheads): + key = f"{reverse_ship_map[name]}_armour_{bulkhead}" + modules[key] = {"mass": m["bulkheads"][i]["mass"]} + + # Native dictionary sorting optimization + ships = dict(sorted(ships.items())) + with open("ships.json", "w", encoding="utf-8") as ships_file: json.dump(ships, ships_file, indent=4) # Module masses - for cat in list(data['Modules'].values()): - for grp, mlist in list(cat.items()): + for cat in data["Modules"].values(): + for grp, mlist in cat.items(): for m in mlist: - if 'symbol' not in m: - raise ValueError(f'No symbol in {m}') - key = str(m['symbol'].lower()) - if grp == 'fsd': + if "symbol" not in m: + raise ValueError(f"No symbol in {m}") + + key = m["symbol"].lower() + if grp == "fsd": modules[key] = { - 'mass': m['mass'], - 'optmass': m['optmass'], - 'maxfuel': m['maxfuel'], - 'fuelmul': m['fuelmul'], - 'fuelpower': m['fuelpower'], + "mass": m["mass"], + "optmass": m["optmass"], + "maxfuel": m["maxfuel"], + "fuelmul": m["fuelmul"], + "fuelpower": m["fuelpower"], } - elif grp == 'gfsb': + elif grp == "gfsb": modules[key] = { - 'mass': m['mass'], - 'jumpboost': m['jumpboost'], + "mass": m["mass"], + "jumpboost": m["jumpboost"], } else: - modules[key] = {'mass': m.get('mass', 0)} # Some modules don't have mass + modules[key] = {"mass": m.get("mass", 0)} # Some modules don't have mass # Pre 3.3 modules - add(modules, 'int_stellarbodydiscoveryscanner_standard', {'mass': 2}) - add(modules, 'int_stellarbodydiscoveryscanner_intermediate', {'mass': 2}) - add(modules, 'int_stellarbodydiscoveryscanner_advanced', {'mass': 2}) + add_module(modules, "int_stellarbodydiscoveryscanner_standard", {"mass": 2}) + add_module(modules, "int_stellarbodydiscoveryscanner_intermediate", {"mass": 2}) + add_module(modules, "int_stellarbodydiscoveryscanner_advanced", {"mass": 2}) - modules = {k: modules[k] for k in sorted(modules)} - with open("modules.json", "w") as modules_file: + modules = dict(sorted(modules.items())) + with open("modules.json", "w", encoding="utf-8") as modules_file: json.dump(modules, modules_file, indent=4) + + +if __name__ == "__main__": + main() diff --git a/killswitch.py b/killswitch.py index af730db0..3e07f94f 100644 --- a/killswitch.py +++ b/killswitch.py @@ -12,7 +12,7 @@ import threading from copy import deepcopy from typing import ( TYPE_CHECKING, Any, NamedTuple, - TypedDict, TypeVar, cast + TypedDict, cast ) from collections.abc import Callable, Mapping, MutableMapping, MutableSequence, Sequence import requests @@ -29,10 +29,8 @@ CURRENT_KILLSWITCH_VERSION = 2 UPDATABLE_DATA = Mapping | Sequence _current_version: semantic_version.Version = config.appversion_nobuild() -T = TypeVar('T', bound=UPDATABLE_DATA) - -class SingleKill(NamedTuple): +class SingleKill[T: Mapping[Any, Any] | Sequence[Any]](NamedTuple): # noqa: D101 """A single KillSwitch. Possibly with additional rules.""" match: str @@ -78,30 +76,27 @@ def _apply(target: UPDATABLE_DATA, key: str, to_set: Any = None, delete: bool = :raises ValueError: when an unexpected target type is passed :raises IndexError: when an invalid index is set or deleted """ - if isinstance(target, MutableMapping): - if delete: - target.pop(key, None) - else: - target[key] = to_set + match target: + case MutableMapping(): + if delete: + target.pop(key, None) + else: + target[key] = to_set + case MutableSequence(): + idx = _get_int(key) + if idx is None: + raise ValueError(f'Cannot use string {key!r} as int for index into Sequence') - elif isinstance(target, MutableSequence): - idx = _get_int(key) - if idx is None: - raise ValueError(f'Cannot use string {key!r} as int for index into Sequence') - - if delete and len(target) > 0: - length = len(target) - if idx in range(-length, length): - target.pop(idx) - - elif len(target) == idx: - target.append(to_set) - - else: - target[idx] = to_set # this can raise, that's fine - - else: - raise ValueError(f'Dont know how to apply data to {type(target)} {target!r}') + if delete and len(target) > 0: + length = len(target) + if idx in range(-length, length): + target.pop(idx) + elif len(target) == idx: + target.append(to_set) + else: + target[idx] = to_set # this can raise, that's fine + case _: + raise ValueError(f'Dont know how to apply data to {type(target)} {target!r}') def _deep_apply(target: UPDATABLE_DATA, path: str, to_set=None, delete=False): # noqa: CCR001 # Recursive silliness. @@ -144,18 +139,17 @@ def _deep_apply(target: UPDATABLE_DATA, path: str, to_set=None, delete=False): else: key, _, path = path.partition('.') - if isinstance(current, Mapping): - current = current[key] # type: ignore # I really don't know at this point what you want from me mypy. - - elif isinstance(current, Sequence): - target_idx = _get_int(key) # mypy is broken. doesn't like := here. - if target_idx is not None: - current = current[target_idx] - else: - raise ValueError(f'Cannot index sequence with non-int key {key!r}') - - else: - raise ValueError(f'Dont know how to index a {type(current)} ({current!r})') + match current: + case Mapping(): + current = current[key] # type: ignore # I really don't know at this point what you want from me mypy. + case Sequence(): + target_idx = _get_int(key) # mypy is broken. doesn't like := here. + if target_idx is not None: + current = current[target_idx] + else: + raise ValueError(f'Cannot index sequence with non-int key {key!r}') + case _: + raise ValueError(f'Dont know how to index a {type(current)} ({current!r})') _apply(current, path, to_set, delete) @@ -212,7 +206,7 @@ class DisabledResult(NamedTuple): return self.has_kill() and self.kill.has_rules # type: ignore -class KillSwitchSet: +class KillSwitchSet[T]: """Queryable set of kill switches.""" def __init__(self, kill_switches: list[KillSwitches]) -> None: @@ -504,12 +498,12 @@ def get_disabled(id: str, *, version: semantic_version.Version = _current_versio return active.get_disabled(id, version=version) -def check_killswitch(name: str, data: T, log=logger) -> tuple[bool, T]: +def check_killswitch[T](name: str, data: T, log=logger) -> tuple[bool, T]: # noqa: D103 """Query the global KillSwitchSet#check_killswitch method.""" return active.check_killswitch(name, data, log) -def check_multiple_killswitches(data: T, *names: str, log=logger) -> tuple[bool, T]: +def check_multiple_killswitches[T](data: T, *names: str, log=logger) -> tuple[bool, T]: # noqa: D103 """Query the global KillSwitchSet#check_multiple method.""" return active.check_multiple_killswitches(data, *names, log=log) diff --git a/loadout.py b/loadout.py index 9d72029d..bd223437 100644 --- a/loadout.py +++ b/loadout.py @@ -1,60 +1,83 @@ -""" -loadout.py - Export ship loadout in Companion API json format. +"""loadout.py - Export ship loadout in Companion API json format. Copyright (c) EDCD, All Rights Reserved Licensed under the GNU General Public License v2 or later. See LICENSE file. """ + from __future__ import annotations import json +import pathlib import re -import time -from pathlib import Path import companion -import util_ships +from datetime import datetime, timezone from config import config from EDMCLogging import get_main_logger +import util_ships logger = get_main_logger() -def export(data: companion.CAPIData, requested_filename: str | None = None) -> None: - """ - Write Ship Loadout in Companion API JSON format. +def export( + data: companion.CAPIData, requested_filename: str | pathlib.Path | None = None +) -> None: + """Write Ship Loadout in Companion API JSON format. :param data: CAPI data containing ship loadout. - :param requested_filename: Name of file to write to. + :param requested_filename: Explicit name or Path of file to write to, or None to auto-generate. """ + # Pretty-Print string = json.dumps( - companion.ship(data), cls=companion.CAPIDataEncoder, - ensure_ascii=False, indent=2, sort_keys=True, separators=(',', ': ') - ) # pretty print + companion.ship(data), + cls=companion.CAPIDataEncoder, + ensure_ascii=False, + indent=2, + sort_keys=True, + separators=(",", ": "), + ) - if requested_filename is not None and requested_filename: - with open(requested_filename, 'w') as h: + if requested_filename: + target_path = pathlib.Path(requested_filename) + with open(target_path, "w", encoding="utf-8") as h: h.write(string) return - if not requested_filename: - logger.error(f"{requested_filename=} is not valid") + + if requested_filename == "": + logger.error("provided requested_filename is an invalid blank string sequence.") return # Look for last ship of this type - ship = util_ships.ship_file_name(data['ship'].get('shipName'), data['ship']['name']) - regexp = re.compile(re.escape(ship) + r'\.\d\d\d\d-\d\d-\d\dT\d\d\.\d\d\.\d\d\.txt') - oldfiles = sorted([x.name for x in Path(config.get_str('outdir')).iterdir() if regexp.match(x.name)]) - if oldfiles: - with open(Path(config.get_str('outdir')) / Path(oldfiles[-1])) as h: - if h.read() == string: - return # same as last time - don't write + ship_data = data.get("ship", {}) + custom_name = ship_data.get("shipName") + raw_ship_type = ship_data.get("name", "UnknownShip") - query_time = config.get_int('querytime', default=int(time.time())) + ship_file_base = util_ships.ship_file_name(custom_name, raw_ship_type) - # Write + regexp = re.compile( + re.escape(ship_file_base) + r"\.\d{4}-\d{2}-\d{2}T\d{2}\.\d{2}\.\d{2}\.txt" + ) - output_directory = Path(config.get_str('outdir')) - ship_time = time.strftime('%Y-%m-%dT%H.%M.%S', time.localtime(query_time)) - file_path = output_directory / f"{ship}.{ship_time}.txt" + out_dir = pathlib.Path(config.get_str("outdir")) - with open(file_path, 'w') as h: + if out_dir.is_dir(): + oldfiles = sorted( + [x for x in out_dir.iterdir() if regexp.match(x.name)], key=lambda p: p.name + ) + + # Deduplication + if oldfiles: + try: + with open(oldfiles[-1], encoding="utf-8") as h: + if h.read() == string: + return + except (OSError, UnicodeDecodeError) as e: + logger.debug(f"Could not read historical loadout log for deduplication: {e!r}") + + querytime = config.get_int("querytime", default=int(datetime.now(timezone.utc).timestamp())) + timestamp = datetime.fromtimestamp(querytime).strftime("%Y-%m-%dT%H.%M.%S") + + final_output_file = out_dir / f"{ship_file_base}.{timestamp}.txt" + + with open(final_output_file, "w", encoding="utf-8") as h: h.write(string) diff --git a/plugins/edastro_core.py b/plugins/edastro_core.py index 17cf1dda..e7059cee 100644 --- a/plugins/edastro_core.py +++ b/plugins/edastro_core.py @@ -19,7 +19,6 @@ referenced in this file (or only in any other core plugin), and if so... IN AN END-USER INSTALLATION ON WINDOWS. """ -import json from typing import Any from tkinter import ttk import requests @@ -34,57 +33,50 @@ from plugins.common_coreutils import PADX, PADY, BUTTONX logger = get_main_logger() +# Module-level globals (Replacing the "This" class architecture) +APP_NAME = "EDAstro" +EDASTRO_PUSH = "https://edastro.com/api/journal" -class This: - """Holds module globals.""" +LOG_ENABLED_VAR: tk.BooleanVar | None = None +LOG_BUTTON: ttk.Checkbutton | None = None - app_name = "EDAstro" - edastro_push = "https://edastro.com/api/journal" - - event_filters = { - "CarrierStats": [ - "timestamp", - "event", - "Callsign", - "Name", - "CarrierID", - "AllowNotorious", - "PendingDecommission", - "DockingAccess", - "FuelLevel", - "Crew", - ], - "CarrierJumpRequest": [ - "timestamp", - "event", - "SystemName", - "SystemAddress", - "CarrierID", - "Body", - "BodyID" - "DepartureTime" - ], - "ScanOrganic": [ - "timestamp", - "ScanType", - "Genus_Localised", - "Species_Localised", - "Genus", - "event", - "Body", - "Species", - "Variant", - "SystemAddress", - "Variant_Localised", - ] - } - - def __init__(self): - self.log: tk.BooleanVar | None = None - self.log_button: ttk.Checkbutton | None = None - - -this = This() +EVENT_FILTERS: dict[str, list[str]] = { + "CarrierStats": [ + "timestamp", + "event", + "Callsign", + "Name", + "CarrierID", + "AllowNotorious", + "PendingDecommission", + "DockingAccess", + "FuelLevel", + "Crew", + ], + "CarrierJumpRequest": [ + "timestamp", + "event", + "SystemName", + "SystemAddress", + "CarrierID", + "Body", + "BodyID", + "DepartureTime", + ], + "ScanOrganic": [ + "timestamp", + "ScanType", + "Genus_Localised", + "Species_Localised", + "Genus", + "event", + "Body", + "Species", + "Variant", + "SystemAddress", + "Variant_Localised", + ] +} def set_config_first_run() -> None: @@ -116,10 +108,11 @@ def plugin_app(parent: tk.Tk) -> None: :param parent: tkinter parent frame. :return: See PLUGINS.md#display """ - this.log = tk.BooleanVar(value=config.get_bool("edastro_send")) + global LOG_ENABLED_VAR + LOG_ENABLED_VAR = tk.BooleanVar(value=config.get_bool("edastro_send")) -def plugin_prefs(parent, cmdr: str, is_beta: bool) -> nb.Frame: +def plugin_prefs(parent: Any, cmdr: str, is_beta: bool) -> nb.Frame: """ Set up Preferences pane for this plugin. @@ -128,6 +121,7 @@ def plugin_prefs(parent, cmdr: str, is_beta: bool) -> nb.Frame: :param is_beta: `bool` - True if this is a beta version of the Game. :return: The tkinter frame we created. """ + global LOG_BUTTON edastroframe = nb.Frame(parent) edastroframe.columnconfigure(0, weight=1) @@ -138,18 +132,18 @@ def plugin_prefs(parent, cmdr: str, is_beta: bool) -> nb.Frame: background=nb.Label().cget("background"), url="https://edastro.com", underline=True, - ).grid( - row=cur_row, padx=PADX, pady=PADY, sticky=tk.W - ) # Don't translate + ).grid(row=cur_row, padx=PADX, pady=PADY, sticky=tk.W) # Don't translate + cur_row += 1 - this.log_button = nb.Checkbutton( + LOG_BUTTON = nb.Checkbutton( edastroframe, # LANG: Settings>EDAstro - Label on checkbox for 'send data' text=tr.tl("Send data to EDAstro"), - variable=this.log, + variable=LOG_ENABLED_VAR, ) - if this.log_button: - this.log_button.grid( + + if LOG_BUTTON: + LOG_BUTTON.grid( row=cur_row, columnspan=2, padx=BUTTONX, pady=PADY, sticky=tk.W ) cur_row += 1 @@ -164,55 +158,57 @@ def prefs_changed(cmdr: str, is_beta: bool) -> None: :param cmdr: Name of Commander. :param is_beta: Whether game beta was detected. """ - if this.log: - config.set("edastro_send", this.log.get()) + if LOG_ENABLED_VAR: + config.set("edastro_send", LOG_ENABLED_VAR.get()) -def filter_event_data(entry) -> dict[str, Any]: - """Format Journal Data for EDAstro.""" - if entry["event"] in this.event_filters: - return { - key: entry[key] - for key in this.event_filters[entry["event"]] - if key in entry - } +def filter_event_data(entry: dict[str, Any]) -> dict[str, Any]: + """Format Journal Data for EDAstro using specific element structures.""" + event_name = entry.get("event") + if event_name in EVENT_FILTERS: + allowed_keys = EVENT_FILTERS[event_name] + return {key: entry[key] for key in allowed_keys if key in entry} return entry -def edastro_update(system, entry, state): +def edastro_update(system: str, entry: dict[str, Any], state: dict[str, Any]) -> None: """Send a processed event to EDAstro.""" - event_name = str(entry["event"]) + event_name = str(entry.get("event", "UnknownEvent")) filtered_entry = filter_event_data(entry) + app_header = { - "appName": this.app_name, + "appName": APP_NAME, "odyssey": state.get("Odyssey"), "system": system, } event_object = [app_header, filtered_entry] - event_data = json.dumps(event_object) + try: - json_header = {"Content-Type": "application/json"} + # Use requests json conversion instead of manual response = requests.post( - url=this.edastro_push, headers=json_header, data=event_data, timeout=20 + url=EDASTRO_PUSH, json=event_object, timeout=20 ) + if response.status_code == 200: - edastro = json.loads(response.text) - if str(edastro["status"]) == "200" or str(edastro["status"]) == "401": + edastro = response.json() + status = edastro.get("status") + + if status in (200, "200", 401, "401"): # 200 = at least one event accepted, 401 = none were accepted, but no errors either logger.info(f"EDAstro: Data sent! ({event_name})") else: logger.debug( - f"Error Response:\nRequest: {this.edastro_push}\n " - f'Response ({edastro["status"]}): \n{edastro["message"]}' + f"Error Response:\nRequest: {EDASTRO_PUSH}\n " + f'Response ({status}): \n{edastro.get("message")}' ) else: logger.debug( - f"Unexpected Response:\nRequest: {this.edastro_push}\n " + f"Unexpected Response:\nRequest: {EDASTRO_PUSH}\n " f"Response ({response.status_code}):\n{response.text}" ) except Exception as ex: logger.warning( - f"Failed to submit EDAstro data:\nRequest: {this.edastro_push}", + f"Failed to submit EDAstro data:\nRequest: {EDASTRO_PUSH}", exc_info=ex, ) @@ -237,8 +233,9 @@ def journal_entry( :return: None if no error, else an error string. """ if ( - this.log is not None and this.log.get() - and entry["event"] in ["CarrierStats", "CarrierJumpRequest", "ScanOrganic"] + LOG_ENABLED_VAR is not None + and LOG_ENABLED_VAR.get() + and entry.get("event") in EVENT_FILTERS ): edastro_update(system, entry, state) diff --git a/plugins/spansh_core.py b/plugins/spansh_core.py index c2795c09..6ad09ccb 100644 --- a/plugins/spansh_core.py +++ b/plugins/spansh_core.py @@ -18,13 +18,14 @@ 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. """ -# pylint: disable=import-error + from __future__ import annotations +import sys import tkinter as tk from tkinter import ttk from typing import Any -import requests +from urllib.parse import quote from companion import CAPIData from config import appname, config from EDMCLogging import get_main_logger @@ -33,24 +34,25 @@ from plugins.common_coreutils import (station_link_common, this_format_common, logger = get_main_logger() +# Module-Level Application State Constants & Variables +IDENTIFIER = "Spansh" -class This: - """Holds module globals.""" +PARENT_ROOT: tk.Tk | None = None +SHUTTING_DOWN: bool = False - def __init__(self): - self.parent: tk.Tk - self.shutting_down = False # Plugin is shutting down. - self.system_link: ttk.Widget = None # type: ignore - self.system_name: str | None = None - self.system_address: str | None = None - self.system_population: int | None = None - self.station_link: ttk.Widget = None # type: ignore - self.station_name = None - self.station_marketid = None - self.on_foot = False +SYSTEM_LINK: ttk.Widget | None = None +SYSTEM_NAME: str | None = None +SYSTEM_ADDRESS: int | str | None = None +SYSTEM_POPULATION: int | None = None +STATION_LINK: ttk.Widget | None = None +STATION_NAME: str | None = None +STATION_MARKETID: int | str | None = None +STATION_TYPE: str | None = None +ON_FOOT: bool = False -this = This() +# Resolve module namespace reference once globally for common_coreutils compatibility +CURRENT_MODULE = sys.modules[__name__] def plugin_start3(plugin_dir: str) -> str: @@ -60,7 +62,7 @@ def plugin_start3(plugin_dir: str) -> str: :param plugin_dir: Name of directory this was loaded from. :return: Identifier string for this plugin. """ - return 'Spansh' + return IDENTIFIER def plugin_app(parent: tk.Tk) -> None: @@ -68,16 +70,20 @@ def plugin_app(parent: tk.Tk) -> None: Construct this plugin's main UI, if any. :param parent: The tk parent to place our widgets into. - :return: See PLUGINS.md#display """ - this.parent = parent - this.system_link = parent.nametowidget(f".{appname.lower()}.system") - this.station_link = parent.nametowidget(f".{appname.lower()}.station") + global PARENT_ROOT, SYSTEM_LINK, STATION_LINK + + PARENT_ROOT = parent + base_path = f".{appname.lower()}" + + SYSTEM_LINK = parent.nametowidget(f"{base_path}.system") + STATION_LINK = parent.nametowidget(f"{base_path}.station") def plugin_stop() -> None: """Plugin shutdown hook.""" - this.shutting_down = True + global SHUTTING_DOWN + SHUTTING_DOWN = True def journal_entry( @@ -94,22 +100,20 @@ def journal_entry( :param state: `monitor.state` :return: None if no error, else an error string. """ - this_format_common(this, state) + this_format_common(CURRENT_MODULE, state) - # Only actually change URLs if we are current provider. - if config.get_str('system_provider') == 'spansh': - this.system_link['text'] = this.system_name - # Do *NOT* set 'url' here, as it's set to a function that will call - # through correctly. We don't want a static string. - this.system_link.update_idletasks() + # Only actually change URLs if we are the designated active provider. + if config.get_str("system_provider") == "spansh" and SYSTEM_LINK: + SYSTEM_LINK["text"] = SYSTEM_NAME + # Do *NOT* set 'url' here, as it's set to a function that will call through correctly. + SYSTEM_LINK.update_idletasks() - if config.get_str('station_provider') == 'spansh': - station_name_setter_common(this) - # Do *NOT* set 'url' here, as it's set to a function that will call - # through correctly. We don't want a static string. - this.station_link.update_idletasks() + if config.get_str("station_provider") == "spansh" and STATION_LINK: + station_name_setter_common(CURRENT_MODULE) + # Do *NOT* set 'url' here, as it's set to a function that will call through correctly. + STATION_LINK.update_idletasks() - return '' + return "" def cmdr_data(data: CAPIData, is_beta: bool) -> str | None: @@ -120,37 +124,34 @@ def cmdr_data(data: CAPIData, is_beta: bool) -> str | None: :param is_beta: Whether game beta was detected. :return: Optional error string. """ - cmdr_data_initial_common(this, data) + cmdr_data_initial_common(CURRENT_MODULE, data) - # Override standard URL functions - if config.get_str('system_provider') == 'spansh': - this.system_link['text'] = this.system_name - # Do *NOT* set 'url' here, as it's set to a function that will call - # through correctly. We don't want a static string. - this.system_link.update_idletasks() - if config.get_str('station_provider') == 'spansh': - station_link_common(data, this) - # Do *NOT* set 'url' here, as it's set to a function that will call - # through correctly. We don't want a static string. - this.station_link.update_idletasks() + # Override standard URL labels if targeted + if config.get_str("system_provider") == "spansh" and SYSTEM_LINK: + SYSTEM_LINK["text"] = SYSTEM_NAME + SYSTEM_LINK.update_idletasks() - return '' + if config.get_str("station_provider") == "spansh" and STATION_LINK: + station_link_common(data, CURRENT_MODULE) + STATION_LINK.update_idletasks() + + return "" def system_url(system_name: str) -> str: """ Construct an appropriate spansh URL for the provided system. - :param system_name: + :param system_name: Name of the star system. :return: The URL, empty if no data was available to construct it. """ if system_name: - return requests.utils.requote_uri(f'https://www.spansh.co.uk/search/{system_name}') + return f"https://www.spansh.co.uk/search/{quote(system_name)}" - if this.system_address: - return requests.utils.requote_uri(f'https://www.spansh.co.uk/system/{this.system_address}') + if SYSTEM_ADDRESS: + return f"https://www.spansh.co.uk/system/{quote(str(SYSTEM_ADDRESS))}" - return '' + return "" def station_url(system_name: str, station_name: str) -> str: @@ -164,7 +165,7 @@ def station_url(system_name: str, station_name: str) -> str: if system_name: return system_url(system_name) - if this.station_marketid: - return requests.utils.requote_uri(f'https://www.spansh.co.uk/station/{this.station_marketid}') + if STATION_MARKETID: + return f"https://www.spansh.co.uk/station/{quote(str(STATION_MARKETID))}" - return '' + return "" diff --git a/td.py b/td.py index 5290c74a..3e4e21dd 100644 --- a/td.py +++ b/td.py @@ -1,56 +1,90 @@ """Export data for Trade Dangerous.""" -import pathlib -from datetime import datetime +from __future__ import annotations + from collections import defaultdict +from datetime import datetime from operator import itemgetter +from pathlib import Path from platform import system +from typing import Any from companion import CAPIData from config import applongname, appversion, config # These are specific to Trade Dangerous, so don't move to edmc_data.py -demandbracketmap = {0: '?', - 1: 'L', - 2: 'M', - 3: 'H', } -stockbracketmap = {0: '-', - 1: 'L', - 2: 'M', - 3: 'H', } +demandbracketmap = { + 0: "?", + 1: "L", + 2: "M", + 3: "H", +} +stockbracketmap = { + 0: "-", + 1: "L", + 2: "M", + 3: "H", +} -def export(data: CAPIData) -> None: +def export(data: CAPIData) -> None: # noqa: CCR001 """Export market data in TD format.""" - data_path = pathlib.Path(config.get_str('outdir')) + data_path = Path(config.get_str("outdir")) # parse timestamp once - ts = datetime.fromisoformat(data['timestamp'].replace('Z', '+00:00')) - ts_file = ts.strftime('%Y-%m-%dT%H.%M.%S') - ts_data = ts.strftime('%Y-%m-%d %H:%M:%S') + if not (raw_timestamp := data.get("timestamp")): + raise ValueError("Missing 'timestamp' in data payload") - system_name = data['lastSystem']['name'].strip() - starport_name = data['lastStarport']['name'].strip() - cmdr_name = data['commander']['name'].strip() + ts = datetime.fromisoformat(raw_timestamp.replace("Z", "+00:00")) + ts_file = ts.strftime("%Y-%m-%dT%H.%M.%S") + ts_data = ts.strftime("%Y-%m-%d %H:%M:%S") + + if not (system_node := data.get("lastSystem")) or not ( + system_name := system_node.get("name", "").strip() + ): + raise KeyError("Missing or empty 'name' key in 'lastSystem'") + + if not (starport_node := data.get("lastStarport")) or not ( + starport_name := starport_node.get("name", "").strip() + ): + raise KeyError("Missing or empty 'name' key in 'lastStarport'") + + if not (cmdr_node := data.get("commander")) or not ( + cmdr_name := cmdr_node.get("name", "").strip() + ): + raise KeyError("Missing or empty 'name' key in 'commander'") data_filename = f"{system_name}.{starport_name}.{ts_file}.prices" - with open(data_path / data_filename, 'w', encoding='utf-8', newline='\n') as h: - h.write('#! trade.py import -\n') - h.write(f'# Created by {applongname} {appversion()} on {system()} for Cmdr {cmdr_name}.\n') - h.write('#\n# \n\n') - h.write(f'@ {system_name}/{starport_name}\n') + with open(data_path / data_filename, "w", encoding="utf-8", newline="\n") as h: + h.write("#! trade.py import -\n") + h.write( + f"# Created by {applongname} {appversion()} on {system()} for Cmdr {cmdr_name}.\n" + ) + h.write( + "#\n# \n\n" + ) + h.write(f"@ {system_name}/{starport_name}\n") # group by category - by_category = defaultdict(list) - for commodity in data['lastStarport']['commodities']: - by_category[commodity['categoryname']].append(commodity) + by_category: dict[str, list[dict[str, Any]]] = defaultdict(list) + for commodity in starport_node.get("commodities", []): + if category_name := commodity.get("categoryname"): + by_category[category_name].append(commodity) for category in sorted(by_category): - h.write(f' + {category}\n') - for commodity in sorted(by_category[category], key=itemgetter('name')): - demand_val = str(int(commodity['demand'])) if commodity['demandBracket'] else '' - stock_val = str(int(commodity['stock'])) if commodity['stockBracket'] else '' + h.write(f" + {category}\n") + for commodity in sorted(by_category[category], key=itemgetter("name")): + demand_val = ( + str(int(commodity["demand"])) + if commodity.get("demandBracket") + else "" + ) + stock_val = ( + str(int(commodity["stock"])) + if commodity.get("stockBracket") + else "" + ) h.write( f" {commodity['name']:<23}" diff --git a/ttkHyperlinkLabel.py b/ttkHyperlinkLabel.py index 0b1f4fa0..672120ad 100644 --- a/ttkHyperlinkLabel.py +++ b/ttkHyperlinkLabel.py @@ -19,11 +19,12 @@ In addition to standard ttk.Label arguments, takes the following arguments: May be imported by plugins """ from __future__ import annotations + import html -from functools import partial import sys import tkinter as tk import webbrowser +from functools import partial from tkinter import font as tk_font from tkinter import ttk from typing import Any @@ -36,11 +37,11 @@ SHIPYARD_HTML_TEMPLATE = """ - + Redirecting you to your {ship_name} at {provider_name}... - + You should be redirected to your {ship_name} at {provider_name} shortly... @@ -97,12 +98,14 @@ class HyperlinkLabel(tk.Label or ttk.Label): # type: ignore target = plug.invoke(url, 'EDSY', 'shipyard_url', loadout, monitor.is_beta) file_name = config.app_dir_path / "last_shipyard.html" - with open(file_name, 'w') as f: - f.write(SHIPYARD_HTML_TEMPLATE.format( + file_name.write_text( + SHIPYARD_HTML_TEMPLATE.format( link=html.escape(str(target)), provider_name=html.escape(str(url)), ship_name=html.escape("Ship") - )) + ), + encoding='utf-8' + ) webbrowser.open(f'file://localhost/{file_name}') @@ -192,33 +195,35 @@ class HyperlinkLabel(tk.Label or ttk.Label): # type: ignore # LANG: Label for 'Copy' as in 'Copy and Paste' menu.add_command(label=tr.tl('Copy'), command=self.copy) # As in Copy and Paste - if self.name == 'ship': - # LANG: Copy the Inara SLEF Format of the active ship to the clipboard - menu.add_command(label=tr.tl('Copy Inara SLEF'), command=self.copy_slef, state=tk.DISABLED) - menu.entryconfigure(1, state=monitor.slef and tk.NORMAL or tk.DISABLED) + match self.name: + case 'ship': + state_val = tk.NORMAL if monitor.slef else tk.DISABLED + # LANG: Copy the Inara SLEF Format of the active ship to the clipboard + menu.add_command(label=tr.tl('Copy Inara SLEF'), command=self.copy_slef, state=tk.DISABLED) + menu.entryconfigure(1, state=state_val) - menu.add_separator() - for url in plug.provides('shipyard_url'): - menu.add_command( - label=tr.tl("Open in {URL}").format(URL=url), # LANG: Open Element In Selected Provider - command=partial(self.open_shipyard, url) - ) + menu.add_separator() + for url in plug.provides('shipyard_url'): + menu.add_command( + label=tr.tl("Open in {URL}").format(URL=url), # LANG: Open Element In Selected Provider + command=partial(self.open_shipyard, url) + ) - if self.name == 'station': - menu.add_separator() - for url in plug.provides('station_url'): - menu.add_command( - label=tr.tl("Open in {URL}").format(URL=url), # LANG: Open Element In Selected Provider - command=partial(self.open_station, url) - ) + case 'station': + menu.add_separator() + for url in plug.provides('station_url'): + menu.add_command( + label=tr.tl("Open in {URL}").format(URL=url), # LANG: Open Element In Selected Provider + command=partial(self.open_station, url) + ) - if self.name == 'system': - menu.add_separator() - for url in plug.provides('system_url'): - menu.add_command( - label=tr.tl("Open in {URL}").format(URL=url), # LANG: Open Element In Selected Provider - command=partial(self.open_system, url) - ) + case 'system': + menu.add_separator() + for url in plug.provides('system_url'): + menu.add_command( + label=tr.tl("Open in {URL}").format(URL=url), # LANG: Open Element In Selected Provider + command=partial(self.open_system, url) + ) if self['text'] and (self.popup_copy(self['text']) if callable(self.popup_copy) else self.popup_copy): menu.post(event.x_root, event.y_root)