ED/EDMarketConnector
1
0
Fork 0
mirror of https://github.com/EDCD/EDMarketConnector.git synced 2025-04-17 17:42:20 +03:00
Code Issues Projects Releases Wiki Activity

plugins/edsm: Lots of type fixing, inc. conditionals where None is possible

Browse Source
This commit is contained in:
Athanasius 2022-12-22 16:35:02 +00:00
parent ca233a40a8
commit 71cbfb8358
No known key found for this signature in database
GPG Key ID: 772697E181BB2767
2 changed files with 146 additions and 111 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
plug.py
View File

@ -7,9 +7,8 @@ import os
import sys import sys
import tkinter as tk import tkinter as tk
from builtins import object, str from builtins import object, str
from typing import Any, Callable, List, Mapping, MutableMapping, Optional, Tuple from tkinter import ttk
from typing import Any, Callable, List, Mapping, MutableMapping, Optional
import ttk
import companion import companion
import myNotebook as nb # noqa: N813 import myNotebook as nb # noqa: N813
@ -200,7 +199,7 @@ def provides(fn_name: str) -> List[str]:
def invoke( def invoke(
plugin_name: str, fallback: str | None, fn_name: str, *args: Tuple plugin_name: str, fallback: str | None, fn_name: str, *args: Any
) -> Optional[str]: ) -> Optional[str]:
""" """
Invoke a function on a named plugin. Invoke a function on a named plugin.

250
plugins/edsm.py
View File

@ -38,12 +38,14 @@ from datetime import datetime, timedelta, timezone
from queue import Queue from queue import Queue
from threading import Thread from threading import Thread
from time import sleep from time import sleep
from typing import TYPE_CHECKING, Any, List, Literal, Mapping, MutableMapping, Optional, Set, Tuple, Union, cast from tkinter import ttk
from typing import TYPE_CHECKING, Any, Dict, List, Literal, Mapping, MutableMapping, Optional, Set, Tuple, Union, cast
import requests import requests
import killswitch import killswitch
import monitor import monitor
import myNotebook
import myNotebook as nb # noqa: N813 import myNotebook as nb # noqa: N813
import plug import plug
from companion import CAPIData from companion import CAPIData
@ -82,8 +84,8 @@ class This:
self.session: requests.Session = requests.Session() self.session: requests.Session = requests.Session()
self.session.headers['User-Agent'] = user_agent self.session.headers['User-Agent'] = user_agent
self.queue: Queue = Queue() # Items to be sent to EDSM by worker thread self.queue: Queue = Queue() # Items to be sent to EDSM by worker thread
self.discarded_events: Set[str] = [] # List discarded events from EDSM self.discarded_events: Set[str] = set() # List discarded events from EDSM
self.lastlookup: requests.Response # Result of last system lookup self.lastlookup: Dict[str, Any] # Result of last system lookup
# Game state # Game state
self.multicrew: bool = False # don't send captain's ship info to EDSM while on a crew self.multicrew: bool = False # don't send captain's ship info to EDSM while on a crew
@ -91,13 +93,13 @@ class This:
self.newgame: bool = False # starting up - batch initial burst of events self.newgame: bool = False # starting up - batch initial burst of events
self.newgame_docked: bool = False # starting up while docked self.newgame_docked: bool = False # starting up while docked
self.navbeaconscan: int = 0 # batch up burst of Scan events after NavBeaconScan self.navbeaconscan: int = 0 # batch up burst of Scan events after NavBeaconScan
self.system_link: tk.Widget = None self.system_link: tk.Widget | None = None
self.system: tk.Tk = None self.system: tk.Tk | None = None
self.system_address: Optional[int] = None # Frontier SystemAddress self.system_address: int | None = None # Frontier SystemAddress
self.system_population: Optional[int] = None self.system_population: int | None = None
self.station_link: tk.Widget = None self.station_link: tk.Widget | None = None
self.station: Optional[str] = None self.station: str | None = None
self.station_marketid: Optional[int] = None # Frontier MarketID self.station_marketid: int | None = None # Frontier MarketID
self.on_foot = False self.on_foot = False
self._IMG_KNOWN = None self._IMG_KNOWN = None
@ -107,19 +109,19 @@ class This:
self.thread: Optional[threading.Thread] = None self.thread: Optional[threading.Thread] = None
self.log = None self.log: tk.IntVar | None = None
self.log_button = None self.log_button: ttk.Checkbutton | None = None
self.label = None self.label: tk.Widget | None = None
self.cmdr_label = None self.cmdr_label: myNotebook.Label | None = None
self.cmdr_text = None self.cmdr_text: myNotebook.Label | None = None
self.user_label = None self.user_label: myNotebook.Label | None = None
self.user = None self.user: myNotebook.Entry | None = None
self.apikey_label = None self.apikey_label: myNotebook.Label | None = None
self.apikey = None self.apikey: myNotebook.Entry | None = None
this = This() this = This()
@ -267,7 +269,7 @@ def plugin_stop() -> None:
logger.debug('Done.') logger.debug('Done.')
def plugin_prefs(parent: tk.Tk, cmdr: str, is_beta: bool) -> tk.Frame: def plugin_prefs(parent: ttk.Notebook, cmdr: str | None, is_beta: bool) -> tk.Frame:
""" """
Plugin preferences setup hook. Plugin preferences setup hook.
@ -300,7 +302,8 @@ def plugin_prefs(parent: tk.Tk, cmdr: str, is_beta: bool) -> tk.Frame:
frame, text=_('Send flight log and Cmdr status to EDSM'), variable=this.log, command=prefsvarchanged frame, text=_('Send flight log and Cmdr status to EDSM'), variable=this.log, command=prefsvarchanged
) )
this.log_button.grid(columnspan=2, padx=BUTTONX, pady=(5, 0), sticky=tk.W) if this.log_button:
this.log_button.grid(columnspan=2, padx=BUTTONX, pady=(5, 0), sticky=tk.W)
nb.Label(frame).grid(sticky=tk.W) # big spacer nb.Label(frame).grid(sticky=tk.W) # big spacer
# Section heading in settings # Section heading in settings
@ -315,61 +318,77 @@ def plugin_prefs(parent: tk.Tk, cmdr: str, is_beta: bool) -> tk.Frame:
cur_row = 10 cur_row = 10
this.label.grid(columnspan=2, padx=PADX, sticky=tk.W) if this.label:
this.label.grid(columnspan=2, padx=PADX, sticky=tk.W)
# LANG: Game Commander name label in EDSM settings if this.cmdr_label and this.cmdr_text:
this.cmdr_label = nb.Label(frame, text=_('Cmdr')) # Main window # LANG: Game Commander name label in EDSM settings
this.cmdr_label.grid(row=cur_row, padx=PADX, sticky=tk.W) this.cmdr_label = nb.Label(frame, text=_('Cmdr')) # Main window
this.cmdr_text = nb.Label(frame) this.cmdr_label.grid(row=cur_row, padx=PADX, sticky=tk.W)
this.cmdr_text.grid(row=cur_row, column=1, padx=PADX, pady=PADY, sticky=tk.W) this.cmdr_text = nb.Label(frame)
this.cmdr_text.grid(row=cur_row, column=1, padx=PADX, pady=PADY, sticky=tk.W)
cur_row += 1 cur_row += 1
# LANG: EDSM Commander name label in EDSM settings if this.user_label and this.label:
this.user_label = nb.Label(frame, text=_('Commander Name')) # EDSM setting # LANG: EDSM Commander name label in EDSM settings
this.user_label.grid(row=cur_row, padx=PADX, sticky=tk.W) this.user_label = nb.Label(frame, text=_('Commander Name')) # EDSM setting
this.user = nb.Entry(frame) this.user_label.grid(row=cur_row, padx=PADX, sticky=tk.W)
this.user.grid(row=cur_row, column=1, padx=PADX, pady=PADY, sticky=tk.EW) this.user = nb.Entry(frame)
this.user.grid(row=cur_row, column=1, padx=PADX, pady=PADY, sticky=tk.EW)
cur_row += 1 cur_row += 1
# LANG: EDSM API key label if this.apikey_label and this.apikey:
this.apikey_label = nb.Label(frame, text=_('API Key')) # EDSM setting # LANG: EDSM API key label
this.apikey_label.grid(row=cur_row, padx=PADX, sticky=tk.W) this.apikey_label = nb.Label(frame, text=_('API Key')) # EDSM setting
this.apikey = nb.Entry(frame) this.apikey_label.grid(row=cur_row, padx=PADX, sticky=tk.W)
this.apikey.grid(row=cur_row, column=1, padx=PADX, pady=PADY, sticky=tk.EW) this.apikey = nb.Entry(frame)
this.apikey.grid(row=cur_row, column=1, padx=PADX, pady=PADY, sticky=tk.EW)
prefs_cmdr_changed(cmdr, is_beta) prefs_cmdr_changed(cmdr, is_beta)
return frame return frame
def prefs_cmdr_changed(cmdr: str, is_beta: bool) -> None: def prefs_cmdr_changed(cmdr: str | None, is_beta: bool) -> None: # noqa: CCR001
""" """
Handle the Commander name changing whilst Settings was open. Handle the Commander name changing whilst Settings was open.
:param cmdr: The new current Commander name. :param cmdr: The new current Commander name.
:param is_beta: Whether game beta was detected. :param is_beta: Whether game beta was detected.
""" """
this.log_button['state'] = tk.NORMAL if cmdr and not is_beta else tk.DISABLED if this.log_button:
this.user['state'] = tk.NORMAL this.log_button['state'] = tk.NORMAL if cmdr and not is_beta else tk.DISABLED
this.user.delete(0, tk.END)
this.apikey['state'] = tk.NORMAL if this.user:
this.apikey.delete(0, tk.END) this.user['state'] = tk.NORMAL
this.user.delete(0, tk.END)
if this.apikey:
this.apikey['state'] = tk.NORMAL
this.apikey.delete(0, tk.END)
if cmdr: if cmdr:
this.cmdr_text['text'] = f'{cmdr}{" [Beta]" if is_beta else ""}' if this.cmdr_text:
this.cmdr_text['text'] = f'{cmdr}{" [Beta]" if is_beta else ""}'
cred = credentials(cmdr) cred = credentials(cmdr)
if cred: if cred:
this.user.insert(0, cred[0]) if this.user:
this.apikey.insert(0, cred[1]) this.user.insert(0, cred[0])
if this.apikey:
this.apikey.insert(0, cred[1])
else: else:
# LANG: We have no data on the current commander if this.cmdr_text:
this.cmdr_text['text'] = _('None') # 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: Union[Literal['normal'], Literal['disabled']] = tk.DISABLED
if cmdr and not is_beta and this.log.get(): if cmdr and not is_beta and this.log and this.log.get():
to_set = tk.NORMAL to_set = tk.NORMAL
set_prefs_ui_states(to_set) set_prefs_ui_states(to_set)
@ -378,7 +397,7 @@ def prefs_cmdr_changed(cmdr: str, is_beta: bool) -> None:
def prefsvarchanged() -> None: def prefsvarchanged() -> None:
"""Handle the 'Send data to EDSM' tickbox changing state.""" """Handle the 'Send data to EDSM' tickbox changing state."""
to_set = tk.DISABLED to_set = tk.DISABLED
if this.log.get(): if this.log and this.log.get() and this.log_button:
to_set = this.log_button['state'] to_set = this.log_button['state']
set_prefs_ui_states(to_set) set_prefs_ui_states(to_set)
@ -390,13 +409,17 @@ def set_prefs_ui_states(state: str) -> None:
:param state: the state to set each entry to :param state: the state to set each entry to
""" """
this.label['state'] = state if (
this.cmdr_label['state'] = state this.label and this.cmdr_label and this.cmdr_text and this.user_label and this.user
this.cmdr_text['state'] = state and this.apikey_label and this.apikey
this.user_label['state'] = state ):
this.user['state'] = state this.label['state'] = state
this.apikey_label['state'] = state this.cmdr_label['state'] = state
this.apikey['state'] = state this.cmdr_text['state'] = state
this.user_label['state'] = state
this.user['state'] = state
this.apikey_label['state'] = state
this.apikey['state'] = state
def prefs_changed(cmdr: str, is_beta: bool) -> None: def prefs_changed(cmdr: str, is_beta: bool) -> None:
@ -406,7 +429,8 @@ def prefs_changed(cmdr: str, is_beta: bool) -> None:
:param cmdr: Name of Commander. :param cmdr: Name of Commander.
:param is_beta: Whether game beta was detected. :param is_beta: Whether game beta was detected.
""" """
config.set('edsm_out', this.log.get()) if this.log:
config.set('edsm_out', this.log.get())
if cmdr and not is_beta: if cmdr and not is_beta:
# TODO: remove this when config is rewritten. # TODO: remove this when config is rewritten.
@ -414,17 +438,18 @@ def prefs_changed(cmdr: str, is_beta: bool) -> None:
usernames: List[str] = config.get_list('edsm_usernames', default=[]) usernames: List[str] = config.get_list('edsm_usernames', default=[])
apikeys: List[str] = config.get_list('edsm_apikeys', default=[]) apikeys: List[str] = config.get_list('edsm_apikeys', default=[])
if cmdr in cmdrs: if this.user and this.apikey:
idx = cmdrs.index(cmdr) if cmdr in cmdrs:
usernames.extend([''] * (1 + idx - len(usernames))) idx = cmdrs.index(cmdr)
usernames[idx] = this.user.get().strip() usernames.extend([''] * (1 + idx - len(usernames)))
apikeys.extend([''] * (1 + idx - len(apikeys))) usernames[idx] = this.user.get().strip()
apikeys[idx] = this.apikey.get().strip() apikeys.extend([''] * (1 + idx - len(apikeys)))
apikeys[idx] = this.apikey.get().strip()
else: else:
config.set('edsm_cmdrs', cmdrs + [cmdr]) config.set('edsm_cmdrs', cmdrs + [cmdr])
usernames.append(this.user.get().strip()) usernames.append(this.user.get().strip())
apikeys.append(this.apikey.get().strip()) apikeys.append(this.apikey.get().strip())
config.set('edsm_usernames', usernames) config.set('edsm_usernames', usernames)
config.set('edsm_apikeys', apikeys) config.set('edsm_apikeys', apikeys)
@ -545,12 +570,13 @@ entry: {entry!r}'''
else: else:
to_set = '' to_set = ''
this.station_link['text'] = to_set if this.station_link:
this.station_link['url'] = station_url(str(this.system), str(this.station)) this.station_link['text'] = to_set
this.station_link.update_idletasks() this.station_link['url'] = station_url(str(this.system), str(this.station))
this.station_link.update_idletasks()
# Update display of 'EDSM Status' image # Update display of 'EDSM Status' image
if this.system_link['text'] != system: if this.system_link and this.system_link['text'] != system:
this.system_link['text'] = system if system else '' this.system_link['text'] = system if system else ''
this.system_link['image'] = '' this.system_link['image'] = ''
this.system_link.update_idletasks() this.system_link.update_idletasks()
@ -639,7 +665,7 @@ Queueing: {entry!r}'''
# Update system data # Update system data
def cmdr_data(data: CAPIData, is_beta: bool) -> Optional[str]: def cmdr_data(data: CAPIData, is_beta: bool) -> Optional[str]: # noqa: CCR001
""" """
Process new CAPI data. Process new CAPI data.
@ -663,27 +689,29 @@ def cmdr_data(data: CAPIData, is_beta: bool) -> Optional[str]:
# TODO: Fire off the EDSM API call to trigger the callback for the icons # TODO: Fire off the EDSM API call to trigger the callback for the icons
if config.get_str('system_provider') == 'EDSM': if config.get_str('system_provider') == 'EDSM':
this.system_link['text'] = this.system if this.system_link:
# Do *NOT* set 'url' here, as it's set to a function that will call this.system_link['text'] = this.system
# through correctly. We don't want a static string. # Do *NOT* set 'url' here, as it's set to a function that will call
this.system_link.update_idletasks() # through correctly. We don't want a static string.
this.system_link.update_idletasks()
if config.get_str('station_provider') == 'EDSM': if config.get_str('station_provider') == 'EDSM':
if data['commander']['docked'] or this.on_foot and this.station: if this.station_link:
this.station_link['text'] = this.station if data['commander']['docked'] or this.on_foot and this.station:
this.station_link['text'] = this.station
elif data['lastStarport']['name'] and data['lastStarport']['name'] != "": elif data['lastStarport']['name'] and data['lastStarport']['name'] != "":
this.station_link['text'] = STATION_UNDOCKED this.station_link['text'] = STATION_UNDOCKED
else: else:
this.station_link['text'] = '' this.station_link['text'] = ''
# Do *NOT* set 'url' here, as it's set to a function that will call # Do *NOT* set 'url' here, as it's set to a function that will call
# through correctly. We don't want a static string. # through correctly. We don't want a static string.
this.station_link.update_idletasks() this.station_link.update_idletasks()
if not this.system_link['text']: if this.system_link and not this.system_link['text']:
this.system_link['text'] = system this.system_link['text'] = system
this.system_link['image'] = '' this.system_link['image'] = ''
this.system_link.update_idletasks() this.system_link.update_idletasks()
@ -759,9 +787,12 @@ def worker() -> None: # noqa: CCR001 C901 # Cant be broken up currently
retrying = 0 retrying = 0
while retrying < 3: while retrying < 3:
if item is None:
item = cast(Tuple[str, str, str, Mapping[str, Any]], ("", {}))
should_skip, new_item = killswitch.check_killswitch( should_skip, new_item = killswitch.check_killswitch(
'plugins.edsm.worker', 'plugins.edsm.worker',
item if item is not None else cast(Tuple[str, Mapping[str, Any]], ("", {})), item,
logger logger
) )
@ -838,8 +869,12 @@ def worker() -> None: # noqa: CCR001 C901 # Cant be broken up currently
if any(p for p in pending if p['event'] in ('CarrierJump', 'FSDJump', 'Location', 'Docked')): if any(p for p in pending if p['event'] in ('CarrierJump', 'FSDJump', 'Location', 'Docked')):
data_elided = data.copy() data_elided = data.copy()
data_elided['apiKey'] = '<elided>' data_elided['apiKey'] = '<elided>'
data_elided['message'] = data_elided['message'].decode('utf-8') if isinstance(data_elided['message'], bytes):
data_elided['commanderName'] = data_elided['commanderName'].decode('utf-8') data_elided['message'] = data_elided['message'].decode('utf-8')
if isinstance(data_elided['commanderName'], bytes):
data_elided['commanderName'] = data_elided['commanderName'].decode('utf-8')
logger.trace_if( logger.trace_if(
'journal.locations', 'journal.locations',
"pending has at least one of ('CarrierJump', 'FSDJump', 'Location', 'Docked')" "pending has at least one of ('CarrierJump', 'FSDJump', 'Location', 'Docked')"
@ -858,11 +893,11 @@ def worker() -> None: # noqa: CCR001 C901 # Cant be broken up currently
'journal.locations', f'Overall POST data (elided) is:\n{json.dumps(data_elided, indent=2)}' 'journal.locations', f'Overall POST data (elided) is:\n{json.dumps(data_elided, indent=2)}'
) )
r = this.session.post(TARGET_URL, data=data, timeout=_TIMEOUT) response = this.session.post(TARGET_URL, data=data, timeout=_TIMEOUT)
logger.trace_if('plugin.edsm.api', f'API response content: {r.content!r}') logger.trace_if('plugin.edsm.api', f'API response content: {response.content!r}')
r.raise_for_status() response.raise_for_status()
reply = r.json() reply = response.json()
msg_num = reply['msgnum'] msg_num = reply['msgnum']
msg = reply['msg'] msg = reply['msg']
# 1xx = OK # 1xx = OK
@ -893,7 +928,7 @@ def worker() -> None: # noqa: CCR001 C901 # Cant be broken up currently
# Update main window's system status # Update main window's system status
this.lastlookup = r this.lastlookup = r
# calls update_status in main thread # calls update_status in main thread
if not config.shutting_down: if not config.shutting_down and this.system_link is not None:
this.system_link.event_generate('<<EDSMStatus>>', when="tail") this.system_link.event_generate('<<EDSMStatus>>', when="tail")
if r['msgnum'] // 100 != 1: # type: ignore if r['msgnum'] // 100 != 1: # type: ignore
@ -996,18 +1031,19 @@ def update_status(event=None) -> None:
# msgnum: 1xx = OK, 2xx = fatal error, 3xx = error, 4xx = ignorable errors. # msgnum: 1xx = OK, 2xx = fatal error, 3xx = error, 4xx = ignorable errors.
def edsm_notify_system(reply: Mapping[str, Any]) -> None: def edsm_notify_system(reply: Mapping[str, Any]) -> None:
"""Update the image next to the system link.""" """Update the image next to the system link."""
if not reply: if this.system_link is not None:
this.system_link['image'] = this._IMG_ERROR if not reply:
# LANG: EDSM Plugin - Error connecting to EDSM API this.system_link['image'] = this._IMG_ERROR
plug.show_error(_("Error: Can't connect to EDSM")) # LANG: EDSM Plugin - Error connecting to EDSM API
plug.show_error(_("Error: Can't connect to EDSM"))
elif reply['msgnum'] // 100 not in (1, 4): elif reply['msgnum'] // 100 not in (1, 4):
this.system_link['image'] = this._IMG_ERROR this.system_link['image'] = this._IMG_ERROR
# LANG: EDSM Plugin - Error message from EDSM API # LANG: EDSM Plugin - Error message from EDSM API
plug.show_error(_('Error: EDSM {MSG}').format(MSG=reply['msg'])) plug.show_error(_('Error: EDSM {MSG}').format(MSG=reply['msg']))
elif reply.get('systemCreated'): elif reply.get('systemCreated'):
this.system_link['image'] = this._IMG_NEW this.system_link['image'] = this._IMG_NEW
else: else:
this.system_link['image'] = this._IMG_KNOWN this.system_link['image'] = this._IMG_KNOWN