Additional Refactors, match case adds

This commit is contained in:
David Sangrey committed 2026-06-08 20:27:26 -04:00
1 parent 1fbd4be5a9
commit 089c17a2f5
7 files changed
+388 -320

No files matched your search

+64 -50
View File
@@ -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()
+36 -42
View File
@@ -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)
+52 -29
View File
@@ -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)
+80 -83
View File
@@ -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)
+57 -56
View File
@@ -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 ""
+64 -30
View File
@@ -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# <item name> <sellCR> <buyCR> <demand> <stock> <timestamp>\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# <item name> <sellCR> <buyCR> <demand> <stock> <timestamp>\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}"
+35 -30
View File
@@ -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 = """
<!DOCTYPE HTML>
<html>
<head>
<meta http-equiv="refresh" content="0; url={link}">
<meta http-equiv=\"refresh\" content=\"0; url={link}\">
<title>Redirecting you to your {ship_name} at {provider_name}...</title>
</head>
<body>
<a href="{link}">
<a href=\"{link}\">
You should be redirected to your {ship_name} at {provider_name} shortly...
</a>
</body>
@@ -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)