diff --git a/EDMarketConnector.py b/EDMarketConnector.py index 8296028ac9..66bea7173d 100755 --- a/EDMarketConnector.py +++ b/EDMarketConnector.py @@ -660,13 +660,13 @@ def update_suit_text(self) -> None: self.suit['text'] = '' return - if (suit := monitor.state.get('SuitCurrent')) is None: + if (suit := monitor.state['SuitCurrent']) is None: self.suit['text'] = f'<{_("Unknown")}>' # LANG: Unknown suit return suitname = suit['edmcName'] - if (suitloadout := monitor.state.get('SuitLoadoutCurrent')) is None: + if (suitloadout := monitor.state['SuitLoadoutCurrent']) is None: self.suit['text'] = '' return @@ -739,7 +739,7 @@ def set_labels(self): """Set main window labels, e.g. after language change.""" self.cmdr_label['text'] = _('Cmdr') + ':' # LANG: Label for commander name in main window # LANG: 'Ship' or multi-crew role label in main window, as applicable - self.ship_label['text'] = (monitor.state['Captain'] and _('Role') or _('Ship')) + ':' # Main window + self.ship_label['text'] = (_('Role') if monitor.state['Captain'] else _('Ship')) + ':' # Main window self.suit_label['text'] = _('Suit') + ':' # LANG: Label for 'Suit' line in main UI self.system_label['text'] = _('System') + ':' # LANG: Label for 'System' line in main UI self.station_label['text'] = _('Station') + ':' # LANG: Label for 'Station' line in main UI @@ -991,7 +991,7 @@ def getandsend(self, event=None, retrying: bool = False): # noqa: C901, CCR001 if monitor.state['Modules']: self.ship.configure(state=True) - if monitor.state.get('SuitCurrent') is not None: + if monitor.state['SuitCurrent'] is not None: if (loadout := data.get('loadout')) is not None: if (suit := loadout.get('suit')) is not None: if (suitname := suit.get('edmcName')) is not None: @@ -1076,6 +1076,7 @@ def crewroletext(role: str) -> str: """ return { None: '', + '': '', 'Idle': '', 'FighterCon': _('Fighter'), # LANG: Multicrew role 'FireCon': _('Gunner'), # LANG: Multicrew role diff --git a/companion.py b/companion.py index 0efae6aa78..72afa40093 100644 --- a/companion.py +++ b/companion.py @@ -28,6 +28,7 @@ from edmc_data import companion_category_map as category_map from EDMCLogging import get_main_logger from monitor import monitor +from monitor_state_dict import SuitDict, SuitLoadoutDict from protocol import protocolhandler logger = get_main_logger() @@ -388,7 +389,7 @@ def authorize(self, payload: str) -> str: # noqa: CCR001 # All 'FID' seen in Journals so far have been 'F' # Frontier, Steam and Epic - if f'F{customer_id}' != monitor.state.get('FID'): + if f'F{customer_id}' != monitor.state['FID']: # LANG: Frontier auth customer_id doesn't match game session FID raise CredentialsError(_("Error: customer_id doesn't match!")) @@ -713,28 +714,37 @@ def suit_update(self, data: CAPIData) -> None: :param data: CAPI data to extra suit data from. """ + current_suit: Optional[SuitDict] if (current_suit := data.get('suit')) is None: # Probably no Odyssey on the account, so point attempting more. return monitor.state['SuitCurrent'] = current_suit # It's easier to always have this in the 'sparse array' dict form - suits = data.get('suits') + suits: Optional[Union[List[SuitDict], Dict[int, SuitDict]]] = data.get('suits') if isinstance(suits, list): monitor.state['Suits'] = dict(enumerate(suits)) - else: + elif isinstance(suits, dict): monitor.state['Suits'] = suits + else: + # If it was neither, it didn't exist, so ... dont muck with it, as something is either broken or + # we didn't see a bunch of info. + pass # We need to be setting our edmcName for all suits - loc_name = monitor.state['SuitCurrent'].get('locName', monitor.state['SuitCurrent']['name']) - monitor.state['SuitCurrent']['edmcName'] = monitor.suit_sane_name(loc_name) + if (current_suit := monitor.state['SuitCurrent']) is not None: + loc_name = current_suit['locName'] if current_suit.get('locName') else current_suit['name'] + current_suit['edmcName'] = monitor.suit_sane_name(loc_name) + for s in monitor.state['Suits']: loc_name = monitor.state['Suits'][s].get('locName', monitor.state['Suits'][s]['name']) monitor.state['Suits'][s]['edmcName'] = monitor.suit_sane_name(loc_name) + suit_loadouts: Optional[Union[List[SuitLoadoutDict], Dict[int, SuitLoadoutDict]]] if (suit_loadouts := data.get('loadouts')) is None: logger.warning('CAPI data had "suit" but no (suit) "loadouts"') + return monitor.state['SuitLoadoutCurrent'] = data.get('loadout') # It's easier to always have this in the 'sparse array' dict form diff --git a/monitor.py b/monitor.py index f1390b1426..ea2657e8d8 100644 --- a/monitor.py +++ b/monitor.py @@ -11,9 +11,16 @@ from os.path import basename, expanduser, isdir, join from sys import platform from time import gmtime, localtime, sleep, strftime, strptime, time -from typing import TYPE_CHECKING, Any, BinaryIO, Dict, List, MutableMapping, Optional +from typing import TYPE_CHECKING, Any, BinaryIO, Dict, List, Literal, MutableMapping, Optional from typing import OrderedDict as OrderedDictT -from typing import Tuple +from typing import Tuple, Union, cast + +from monitor_state_dict import ( + ModuleDict, ModuleEngineering, MonitorStateDict, NavRouteDict, OdysseyWeapon, SuitDict, SuitLoadoutDict +) + +# spell-checker: words loadoutid slotid fdev fid relog onfoot fsdjump cheaty suitid fauto sauto intimidator navroute +# spell-checker: words quitacrew joinacrew sellshiponrebuy npccrewpaidwage if TYPE_CHECKING: import tkinter @@ -115,17 +122,17 @@ def __init__(self) -> None: def __init_state(self) -> None: # Cmdr state shared with EDSM and plugins # If you change anything here update PLUGINS.md documentation! - self.state: Dict = { - 'GameLanguage': None, # From `Fileheader - 'GameVersion': None, # From `Fileheader - 'GameBuild': None, # From `Fileheader - 'Captain': None, # On a crew + self.state: MonitorStateDict = { + 'GameLanguage': '', # From `Fileheader + 'GameVersion': '', # From `Fileheader + 'GameBuild': '', # From `Fileheader + 'Captain': '', # On a crew 'Cargo': defaultdict(int), - 'Credits': None, - 'FID': None, # Frontier Cmdr ID - 'Horizons': None, # Does this user have Horizons? - 'Odyssey': False, # Have we detected we're running under Odyssey? - 'Loan': None, + 'Credits': 0, + 'FID': '', # Frontier Cmdr ID + 'Horizons': False, # Does this user have Horizons? + 'Odyssey': False, # Have we detected we're running under Odyssey? + 'Loan': 0, 'Raw': defaultdict(int), 'Manufactured': defaultdict(int), 'Encoded': defaultdict(int), @@ -133,39 +140,40 @@ def __init_state(self) -> None: 'Rank': {}, 'Reputation': {}, 'Statistics': {}, - 'Role': None, # Crew role - None, Idle, FireCon, FighterCon - 'Friends': set(), # Online friends - 'ShipID': None, - 'ShipIdent': None, - 'ShipName': None, - 'ShipType': None, - 'HullValue': None, - 'ModulesValue': None, - 'Rebuy': None, - 'Modules': None, - 'CargoJSON': None, # The raw data from the last time cargo.json was read - 'Route': None, # Last plotted route from Route.json file - 'OnFoot': False, # Whether we think you're on-foot - 'Component': defaultdict(int), # Odyssey Components in Ship Locker - 'Item': defaultdict(int), # Odyssey Items in Ship Locker - 'Consumable': defaultdict(int), # Odyssey Consumables in Ship Locker - 'Data': defaultdict(int), # Odyssey Data in Ship Locker - 'BackPack': { # Odyssey BackPack contents - 'Component': defaultdict(int), # BackPack Components - 'Consumable': defaultdict(int), # BackPack Consumables - 'Item': defaultdict(int), # BackPack Items - 'Data': defaultdict(int), # Backpack Data + 'Role': '', # Crew role - None, Idle, FireCon, FighterCon + 'Friends': set(), # Online friends + 'ShipID': 0-1, # WORKAROUND: https://github.com/PyCQA/pycodestyle/issues/1008 + 'ShipIdent': '', + 'ShipName': '', + 'ShipType': '', + 'HullValue': 0, + 'ModulesValue': 0, + 'Rebuy': 0, + 'Modules': {}, + 'CargoJSON': {}, # The raw data from the last time cargo.json was read + 'NavRoute': NavRouteDict(timestamp='', route=[]), # Last plotted route from Route.json file + 'OnFoot': False, # Whether we think you're on-foot + 'Component': defaultdict(int), # Odyssey Components in Ship Locker + 'Item': defaultdict(int), # Odyssey Items in Ship Locker + 'Consumable': defaultdict(int), # Odyssey Consumables in Ship Locker + 'Data': defaultdict(int), # Odyssey Data in Ship Locker + 'BackPack': { # Odyssey BackPack contents + 'Component': defaultdict(int), # BackPack Components + 'Consumable': defaultdict(int), # BackPack Consumables + 'Item': defaultdict(int), # BackPack Items + 'Data': defaultdict(int), # Backpack Data }, - 'BackpackJSON': None, # Raw JSON from `Backpack.json` file, if available - 'ShipLockerJSON': None, # Raw JSON from the `ShipLocker.json` file, if available + 'BackpackJSON': {}, # Raw JSON from `Backpack.json` file, if available + 'ShipLockerJSON': {}, # Raw JSON from the `ShipLocker.json` file, if available 'SuitCurrent': None, 'Suits': {}, 'SuitLoadoutCurrent': None, 'SuitLoadouts': {}, - 'Taxi': None, # True whenever we are _in_ a taxi. ie, this is reset on Disembark etc. - 'Dropship': None, # Best effort as to whether or not the above taxi is a dropship. - 'Body': None, - 'BodyType': None, + 'Taxi': False, # True whenever we are _in_ a taxi. ie, this is reset on Disembark etc. + 'Dropship': False, # Best effort as to whether or not the above taxi is a dropship. + 'Body': '', + 'BodyType': '', + 'ModuleInfo': {}, } def start(self, root: 'tkinter.Tk') -> bool: # noqa: CCR001 @@ -265,8 +273,8 @@ def stop(self) -> None: self.systemaddress = None self.is_beta = False self.state['OnFoot'] = False - self.state['Body'] = None - self.state['BodyType'] = None + self.state['Body'] = '' + self.state['BodyType'] = '' if self.observed: logger.debug('self.observed: Calling unschedule_all()') @@ -383,7 +391,8 @@ def worker(self) -> None: # noqa: C901, CCR001 # Watchdog thread -- there is a way to get this by using self.observer.emitters and checking for an attribute: # watch, but that may have unforseen differences in behaviour. - emitter = self.observed and self.observer._emitter_for_watch[self.observed] # Note: Uses undocumented attribute + # HACK: Uses undocumented attribute + emitter = self.observed and self.observer._emitter_for_watch[self.observed] # type: ignore logger.debug('Entering loop...') while True: @@ -541,23 +550,24 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C self.systemaddress = None self.started = timegm(strptime(entry['timestamp'], '%Y-%m-%dT%H:%M:%SZ')) # Don't set Ship, ShipID etc since this will reflect Fighter or SRV if starting in those - self.state.update({ - 'Captain': None, - 'Credits': entry['Credits'], - 'FID': entry.get('FID'), # From 3.3 - 'Horizons': entry['Horizons'], # From 3.0 - 'Odyssey': entry.get('Odyssey', False), # From 4.0 Odyssey - 'Loan': entry['Loan'], - 'Engineers': {}, - 'Rank': {}, - 'Reputation': {}, - 'Statistics': {}, - 'Role': None, - 'Taxi': None, - 'Dropship': None, - 'Body': None, - 'BodyType': None, - }) + + # Cant use update() without the entire thing, do stuff manually here + self.state['Captain'] = '' + self.state['Credits'] = entry['Credits'] + self.state['FID'] = entry.get('FID', '') # From 3.3 + self.state['Horizons'] = entry['Horizons'] # From 3.0 + self.state['Odyssey'] = entry.get('Odyssey', False) # From 4.0 Odyssey + self.state['Loan'] = entry['Loan'] + self.state['Engineers'] = {} + self.state['Rank'] = {} + self.state['Reputation'] = {} + self.state['Statistics'] = {} + self.state['Role'] = '' + self.state['Taxi'] = False + self.state['Dropship'] = False + self.state['Body'] = '' + self.state['BodyType'] = '' + if entry.get('Ship') is not None and self._RE_SHIP_ONFOOT.search(entry['Ship']): self.state['OnFoot'] = True @@ -573,30 +583,30 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C if 'UserShipId' in entry: # Only present when changing the ship's ident self.state['ShipIdent'] = entry['UserShipId'] - self.state['ShipName'] = entry.get('UserShipName') + self.state['ShipName'] = entry.get('UserShipName', '') self.state['ShipType'] = self.canonicalise(entry['Ship']) elif event_type == 'shipyardbuy': - self.state['ShipID'] = None - self.state['ShipIdent'] = None - self.state['ShipName'] = None + self.state['ShipID'] = -1 + self.state['ShipIdent'] = '' + self.state['ShipName'] = '' self.state['ShipType'] = self.canonicalise(entry['ShipType']) - self.state['HullValue'] = None - self.state['ModulesValue'] = None - self.state['Rebuy'] = None - self.state['Modules'] = None + self.state['HullValue'] = 0 + self.state['ModulesValue'] = 0 + self.state['Rebuy'] = 0 + self.state['Modules'] = {} self.state['Credits'] -= entry.get('ShipPrice', 0) elif event_type == 'shipyardswap': self.state['ShipID'] = entry['ShipID'] - self.state['ShipIdent'] = None - self.state['ShipName'] = None + self.state['ShipIdent'] = '' + self.state['ShipName'] = '' self.state['ShipType'] = self.canonicalise(entry['ShipType']) - self.state['HullValue'] = None - self.state['ModulesValue'] = None - self.state['Rebuy'] = None - self.state['Modules'] = None + self.state['HullValue'] = 0 + self.state['ModulesValue'] = 0 + self.state['Rebuy'] = 0 + self.state['Modules'] = {} elif ( event_type == 'loadout' and @@ -614,9 +624,9 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C self.state['ShipName'] = entry['ShipName'] self.state['ShipType'] = self.canonicalise(entry['Ship']) - self.state['HullValue'] = entry.get('HullValue') # not present on exiting Outfitting - self.state['ModulesValue'] = entry.get('ModulesValue') # not present on exiting Outfitting - self.state['Rebuy'] = entry.get('Rebuy') + self.state['HullValue'] = entry.get('HullValue', 0) # not present on exiting Outfitting + self.state['ModulesValue'] = entry.get('ModulesValue', 0) # not present on exiting Outfitting + self.state['Rebuy'] = entry.get('Rebuy', 0) # Remove spurious differences between initial Loadout event and subsequent self.state['Modules'] = {} for module in entry['Modules']: @@ -631,7 +641,7 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C self.state['Modules'][module['Slot']] = module elif event_type == 'modulebuy': - self.state['Modules'][entry['Slot']] = { + new_module: ModuleDict = { 'Slot': entry['Slot'], 'Item': self.canonicalise(entry['BuyItem']), 'On': True, @@ -639,6 +649,7 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C 'Health': 1.0, 'Value': entry['BuyPrice'], } + self.state['Modules'][entry['Slot']] = new_module self.state['Credits'] -= entry.get('BuyPrice', 0) @@ -678,7 +689,7 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C # This event is logged when a player (on foot) gets into a ship or SRV # Parameters: # • SRV: true if getting into SRV, false if getting into a ship - # • Taxi: true when boarding a taxi transposrt ship + # • Taxi: true when boarding a taxi transport ship # • Multicrew: true when boarding another player’s vessel # • ID: player’s ship ID (if players own vessel) # • StarSystem @@ -706,7 +717,7 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C # # Parameters: # • SRV: true if getting out of SRV, false if getting out of a ship - # • Taxi: true when getting out of a taxi transposrt ship + # • Taxi: true when getting out of a taxi transport ship # • Multicrew: true when getting out of another player’s vessel # • ID: player’s ship ID (if players own vessel) # • StarSystem @@ -756,16 +767,16 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C # • OnFoot: bool if event_type in ('location', 'carrierjump'): self.planet = entry.get('Body') if entry.get('BodyType') == 'Planet' else None - self.state['Body'] = entry.get('Body') - self.state['BodyType'] = entry.get('BodyType') + self.state['Body'] = entry.get('Body', '') + self.state['BodyType'] = entry.get('BodyType', '') # if event_type == 'location': # logger.trace('"Location" event') elif event_type == 'fsdjump': self.planet = None - self.state['Body'] = None - self.state['BodyType'] = None + self.state['Body'] = '' + self.state['BodyType'] = '' if 'StarPos' in entry: self.coordinates = tuple(entry['StarPos']) # type: ignore @@ -787,9 +798,9 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C self.stationtype = entry.get('StationType') # May be None self.stationservices = entry.get('StationServices') # None in Odyssey for on-foot 'Location' - self.state['Taxi'] = entry.get('Taxi', None) + self.state['Taxi'] = entry.get('Taxi', False) if not self.state['Taxi']: - self.state['Dropship'] = None + self.state['Dropship'] = False elif event_type == 'approachbody': self.planet = entry['Body'] @@ -798,8 +809,8 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C elif event_type in ('leavebody', 'supercruiseentry'): self.planet = None - self.state['Body'] = None - self.state['BodyType'] = None + self.state['Body'] = '' + self.state['BodyType'] = '' elif event_type in ('rank', 'promotion'): payload = dict(entry) @@ -820,7 +831,7 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C payload.pop('event') payload.pop('timestamp') # NB: We need the original casing for these keys - self.state[entry['event']] = payload + self.state[entry['event']] = payload # type: ignore # Non-literal, but the options are ensured above elif event_type == 'engineerprogress': # Sanity check - at least once the 'Engineer' (name) was missing from this in early @@ -829,10 +840,11 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C if self.event_valid_engineerprogress(entry): engineers = self.state['Engineers'] if 'Engineers' in entry: # Startup summary - self.state['Engineers'] = { + to_set: Dict[str, Union[str, Tuple[int, int]]] = { e['Engineer']: ((e['Rank'], e.get('RankProgress', 0)) if 'Rank' in e else e['Progress']) for e in entry['Engineers'] } + self.state['Engineers'] = to_set else: # Promotion engineer = entry['Engineer'] @@ -847,7 +859,7 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C # From 3.3 full Cargo event (after the first one) is written to a separate file if 'Inventory' not in entry: with open(join(self.currentdir, 'Cargo.json'), 'rb') as h: # type: ignore - entry = json.load(h, object_pairs_hook=OrderedDict) # Preserve property order because why not? + entry = json.load(h) self.state['CargoJSON'] = entry clean = self.coalesce_cargo(entry['Inventory']) @@ -1002,6 +1014,8 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C for c in entry[changes]: category = self.category(c['Type']) name = self.canonicalise(c['Name']) + # Cheaty "its fine I promise" for TypedDict + category = cast(Literal['Component', 'Data', 'Consumable', 'Item'], category) if changes == 'Removed': self.state['BackPack'][category][name] -= c['Count'] @@ -1013,9 +1027,9 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C # As of Odyssey Alpha Phase 1 Hotfix 2 keeping track of BackPack # materials is impossible when used/picked up anyway. for c in self.state['BackPack']: - for m in self.state['BackPack'][c]: - if self.state['BackPack'][c][m] < 0: - self.state['BackPack'][c][m] = 0 + for m in self.state['BackPack'][c]: # type: ignore # c and m are dynamic but "safe" + if self.state['BackPack'][c][m] < 0: # type: ignore # c and m are dynamic but "safe" + self.state['BackPack'][c][m] = 0 # type: ignore # c and m are dynamic but "safe" elif event_type == 'buymicroresources': # From 4.0.0.400 we get an empty (see file) `ShipLocker` event, @@ -1089,7 +1103,7 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C if self.state['SuitLoadouts']: loadout_id = self.suit_loadout_id_from_loadoutid(entry['LoadoutID']) try: - self.state['SuitLoadouts'].pop(f'{loadout_id}') + self.state['SuitLoadouts'].pop(loadout_id) except KeyError: # This should no longer happen, as we're now handling CreateSuitLoadout properly @@ -1119,7 +1133,7 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C # { "timestamp":"2021-04-29T09:03:37Z", "event":"BuySuit", "Name":"UtilitySuit_Class1", # "Name_Localised":"Maverick Suit", "Price":150000, "SuitID":1698364934364699 } loc_name = entry.get('Name_Localised', entry['Name']) - self.state['Suits'][entry['SuitID']] = { + to_set_suit: SuitDict = { 'name': entry['Name'], 'locName': loc_name, 'edmcName': self.suit_sane_name(loc_name), @@ -1127,6 +1141,7 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C 'suitId': entry['SuitID'], 'mods': entry['SuitMods'], # Suits can (rarely) be bought with modules installed } + self.state['Suits'][entry['SuitID']] = to_set_suit # update credits if price := entry.get('Price') is None: @@ -1184,16 +1199,18 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C if self.state['SuitLoadouts']: loadout_id = self.suit_loadout_id_from_loadoutid(entry['LoadoutID']) try: - self.state['SuitLoadouts'][loadout_id]['slots'][entry['SlotName']] = { + w_to_set: OdysseyWeapon = { 'name': entry['ModuleName'], 'locName': entry.get('ModuleName_Localised', entry['ModuleName']), 'id': None, 'weaponrackId': entry['SuitModuleID'], 'locDescription': '', 'class': entry['Class'], - 'mods': entry['WeaponMods'] + 'mods': entry['WeaponMods'], } + self.state['SuitLoadouts'][loadout_id]['slots'][entry['SlotName']] = w_to_set + except KeyError: logger.error(f"LoadoutEquipModule: {entry}") @@ -1290,24 +1307,25 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C # Added in ED 3.7 - multi-hop route details in NavRoute.json with open(join(self.currentdir, 'NavRoute.json'), 'rb') as rf: # type: ignore try: - entry = json.load(rf) + nv_entry: NavRouteDict = json.load(rf) except json.JSONDecodeError: logger.exception('Failed decoding NavRoute.json', exc_info=True) else: - self.state['NavRoute'] = entry + self.state['NavRoute'] = nv_entry + entry = cast(dict, nv_entry) elif event_type == 'moduleinfo': with open(join(self.currentdir, 'ModulesInfo.json'), 'rb') as mf: # type: ignore try: - entry = json.load(mf) + m_entry = json.load(mf) except json.JSONDecodeError: logger.exception('Failed decoding ModulesInfo.json', exc_info=True) else: - self.state['ModuleInfo'] = entry + self.state['ModuleInfo'] = m_entry elif event_type in ('collectcargo', 'marketbuy', 'buydrones', 'miningrefined'): commodity = self.canonicalise(entry['Type']) @@ -1342,6 +1360,7 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C elif event_type == 'materials': for category in ('Raw', 'Manufactured', 'Encoded'): + category = cast(Literal['Raw', 'Manufactured', 'Encoded'], category) self.state[category] = defaultdict(int) self.state[category].update({ self.canonicalise(x['Name']): x['Count'] for x in entry.get(category, []) @@ -1349,17 +1368,18 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C elif event_type == 'materialcollected': material = self.canonicalise(entry['Name']) - self.state[entry['Category']][material] += entry['Count'] + self.state[entry['Category']][material] += entry['Count'] # type: ignore elif event_type in ('materialdiscarded', 'scientificresearch'): material = self.canonicalise(entry['Name']) - state_category = self.state[entry['Category']] + state_category = self.state[entry['Category']] # type: ignore state_category[material] -= entry['Count'] if state_category[material] <= 0: state_category.pop(material) elif event_type == 'synthesis': for category in ('Raw', 'Manufactured', 'Encoded'): + category = cast(Literal['Raw', 'Manufactured', 'Encoded'], category) for x in entry['Materials']: material = self.canonicalise(x['Name']) if material in self.state[category]: @@ -1369,7 +1389,7 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C elif event_type == 'materialtrade': category = self.category(entry['Paid']['Category']) - state_category = self.state[category] + state_category = self.state[category] # type: ignore paid = entry['Paid'] received = entry['Received'] @@ -1385,6 +1405,8 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C ): for category in ('Raw', 'Manufactured', 'Encoded'): + category = cast(Literal['Raw', 'Manufactured', 'Encoded'], category) + for x in entry.get('Ingredients', []): material = self.canonicalise(x['Name']) if material in self.state[category]: @@ -1394,7 +1416,7 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C module = self.state['Modules'][entry['Slot']] assert(module['Item'] == self.canonicalise(entry['Module'])) - module['Engineering'] = { + to_set_me: ModuleEngineering = { 'Engineer': entry['Engineer'], 'EngineerID': entry['EngineerID'], 'BlueprintName': entry['BlueprintName'], @@ -1404,6 +1426,8 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C 'Modifiers': entry['Modifiers'], } + module['Engineering'] = to_set_me + if 'ExperimentalEffect' in entry: module['Engineering']['ExperimentalEffect'] = entry['ExperimentalEffect'] module['Engineering']['ExperimentalEffect_Localised'] = entry['ExperimentalEffect_Localised'] @@ -1423,7 +1447,7 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C if 'Category' in reward: # Category not present in E:D 3.0 category = self.category(reward['Category']) material = self.canonicalise(reward['Name']) - self.state[category][material] += reward.get('Count', 1) + self.state[category][material] += reward.get('Count', 1) # type: ignore elif event_type == 'engineercontribution': commodity = self.canonicalise(entry.get('Commodity')) @@ -1435,6 +1459,7 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C material = self.canonicalise(entry.get('Material')) if material: for category in ('Raw', 'Manufactured', 'Encoded'): + category = cast(Literal['Raw', 'Manufactured', 'Encoded'], category) if material in self.state[category]: self.state[category][material] -= entry['Quantity'] if self.state[category][material] <= 0: @@ -1443,6 +1468,7 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C elif event_type == 'technologybroker': for thing in entry.get('Ingredients', []): # 3.01 for category in ('Cargo', 'Raw', 'Manufactured', 'Encoded'): + category = cast(Literal['Raw', 'Manufactured', 'Encoded'], category) item = self.canonicalise(thing['Name']) if item in self.state[category]: self.state[category][item] -= thing['Count'] @@ -1458,6 +1484,7 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C for thing in entry.get('Materials', []): # 3.02 material = self.canonicalise(thing['Name']) category = thing['Category'] + category = cast(Literal['Raw', 'Manufactured', 'Encoded'], category) self.state[category][material] -= thing['Count'] if self.state[category][material] <= 0: self.state[category].pop(material) @@ -1475,15 +1502,15 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C self.systemaddress = None self.state['OnFoot'] = False - self.state['Body'] = None - self.state['BodyType'] = None + self.state['Body'] = '' + self.state['BodyType'] = '' elif event_type == 'changecrewrole': self.state['Role'] = entry['Role'] elif event_type == 'quitacrew': - self.state['Captain'] = None - self.state['Role'] = None + self.state['Captain'] = '' + self.state['Role'] = '' self.planet = None self.system = None self.station = None @@ -1493,8 +1520,8 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C self.coordinates = None self.systemaddress = None - self.state['Body'] = None - self.state['BodyType'] = None + self.state['Body'] = '' + self.state['BodyType'] = '' # TODO: on_foot: Will we get an event after this to know ? elif event_type == 'friends': @@ -1563,8 +1590,8 @@ def parse_entry(self, line: bytes) -> MutableMapping[str, Any]: # noqa: C901, C self.state['Credits'] -= entry.get('Price', 0) elif event_type == 'carrierbanktransfer': - if (newbal := entry.get('PlayerBalance')): - self.state['Credits'] = newbal + if (new_bal := entry.get('PlayerBalance')): + self.state['Credits'] = new_bal elif event_type == 'carrierdecommission': # v30 doc says nothing about citing the refund amount @@ -1668,7 +1695,7 @@ def suitloadout_store_from_event(self, entry) -> Tuple[int, int]: # Check if this looks like a suit we already have stored, so as # to avoid 'bad' Journal localised names. - suit = self.state['Suits'].get(f"{suitid}", None) + suit: Optional[SuitDict] = self.state['Suits'].get(suitid, None) if suit is None: # Initial suit containing just the data that is then embedded in # the loadout @@ -1678,8 +1705,12 @@ def suitloadout_store_from_event(self, entry) -> Tuple[int, int]: suitname = entry.get('SuitName_Localised', entry['SuitName']) edmc_suitname = self.suit_sane_name(suitname) suit = { - 'edmcName': edmc_suitname, - 'locName': suitname, + 'edmcName': edmc_suitname, + 'locName': suitname, + 'suitId': 0-1, # WORKAROUND: https://github.com/PyCQA/pycodestyle/issues/1008 + 'id': None, + 'name': entry['SuitName'], + 'mods': entry['SuitMods'] } # Overwrite with latest data, just in case, as this can be from CAPI which may or may not have had @@ -1690,19 +1721,20 @@ def suitloadout_store_from_event(self, entry) -> Tuple[int, int]: suitloadout_slotid = self.suit_loadout_id_from_loadoutid(entry['LoadoutID']) # Make the new loadout, in the CAPI format - new_loadout = { + new_loadout: SuitLoadoutDict = { 'loadoutSlotId': suitloadout_slotid, 'suit': suit, 'name': entry['LoadoutName'], 'slots': self.suit_loadout_slots_array_to_dict(entry['Modules']), } + # Assign this loadout into our state - self.state['SuitLoadouts'][f"{suitloadout_slotid}"] = new_loadout + self.state['SuitLoadouts'][suitloadout_slotid] = new_loadout # Now add in the extra fields for new_suit to be a 'full' Suit structure suit['id'] = suit.get('id') # Not available in 4.0.0.100 journal event # Ensure the suit is in self.state['Suits'] - self.state['Suits'][f"{suitid}"] = suit + self.state['Suits'][suitid] = suit return suitid, suitloadout_slotid @@ -1717,17 +1749,14 @@ def suit_and_loadout_setcurrent(self, suitid: int, suitloadout_slotid: int) -> b :param suitloadout_slotid: Numeric ID of the slot for the suit loadout. :return: True if we could do this, False if not. """ - str_suitid = f"{suitid}" - str_suitloadoutid = f"{suitloadout_slotid}" - - if (self.state['Suits'].get(str_suitid, False) - and self.state['SuitLoadouts'].get(str_suitloadoutid, False)): - self.state['SuitCurrent'] = self.state['Suits'][str_suitid] - self.state['SuitLoadoutCurrent'] = self.state['SuitLoadouts'][str_suitloadoutid] + if suitid in self.state['Suits'] and suitloadout_slotid in self.state['SuitLoadouts']: + self.state['SuitCurrent'] = self.state['Suits'][suitid] + self.state['SuitLoadoutCurrent'] = self.state['SuitLoadouts'][suitloadout_slotid] return True logger.error(f"Tried to set a suit and suitloadout where we didn't know about both: {suitid=}, " - f"{str_suitloadoutid=}") + f"{suitloadout_slotid=}") + return False # TODO: *This* will need refactoring and a proper validation infrastructure diff --git a/monitor_state_dict.py b/monitor_state_dict.py new file mode 100644 index 0000000000..3d6d180a3b --- /dev/null +++ b/monitor_state_dict.py @@ -0,0 +1,191 @@ +""" +Contains the definitions for monitor.state. + +This is essentially a stopgap while OOP state is worked on. +""" +from __future__ import annotations + +from typing import Any, DefaultDict, Dict, List, MutableMapping, Optional, Set, Tuple, TypedDict, Union + + +class MonitorStateDict(TypedDict): + """Top level state dictionary for monitor.py.""" + + # Game related + GameLanguage: str # From `Fileheader` + GameVersion: str # From `Fileheader` + GameBuild: str # From `Fileheader` + Horizons: bool # Does the player have Horizons? + Odyssey: bool # Have we detected Odyssey? + + # Multi-crew + + Captain: str # If on a crew, the captian's name + Role: str # Role in crew + + # Cmdr state + FID: str # Frontier CMDR ID + Friends: Set[str] # Online Friends + Credits: int + Loan: int + + # (A_D) One day I will change this to be a NamedTuple. But for now it will suffice to state that: + # (Rank, RankProgress) | Literal['Known', 'Invited' (or any other of the possible states that ISNT a rank number)] + Engineers: Dict[str, Union[str, Tuple[int, int]]] + Rank: Dict[str, Tuple[int, int]] # (RankMajor, RankProgress) + Reputation: Dict[str, float] # Superpower -> level + Statistics: Dict[Any, Any] # This is very freeform. + + # Engineering Materials + Raw: DefaultDict[str, int] + Encoded: DefaultDict[str, int] + Manufactured: DefaultDict[str, int] + + # Ship + ShipID: int + ShipIdent: str + ShipName: str + ShipType: str + + HullValue: int + ModulesValue: int + Rebuy: int + Modules: Dict[str, ModuleDict] + ModuleInfo: MutableMapping[Any, Any] # From the game, freeform + + # Cargo (yes technically its on the cmdr not the ship but this makes more sense.) + CargoJSON: MutableMapping[str, Any] # Raw data from the last cargo.json read + Cargo: DefaultDict[str, int] + + # Navigation + NavRoute: NavRouteDict # Last route plotted + Body: str + BodyType: str + Taxi: bool + Dropship: bool + + # Odyssey + OnFoot: bool + Component: DefaultDict[str, int] + Item: DefaultDict[str, int] + Consumable: DefaultDict[str, int] + Data: DefaultDict[str, int] + BackPack: OdysseyBackpack + BackpackJSON: MutableMapping[str, Any] # Direct from Game + ShipLockerJSON: MutableMapping[str, Any] # Direct from Game + + SuitCurrent: Optional[SuitDict] + Suits: Dict[int, SuitDict] + SuitLoadoutCurrent: Optional[SuitLoadoutDict] + SuitLoadouts: Dict[int, SuitLoadoutDict] + + +class OdysseyBackpack(TypedDict): + """Odyssey Backpack contents (used when on-foot).""" + + Component: DefaultDict[str, int] + Item: DefaultDict[str, int] + Consumable: DefaultDict[str, int] + Data: DefaultDict[str, int] + + +class NavRouteDict(TypedDict): + """Description of navroute.json at time of writing.""" + + timestamp: str + route: List[NavRouteEntry] + + +class NavRouteEntry(TypedDict): + """Single NavRoute entry.""" + + StarSystem: str + SystemAddress: int + StarPos: Tuple[float, float, float] + StarClass: str + + +class SuitLoadoutDict(TypedDict): + """Single suit loadout.""" + + loadoutSlotId: int # noqa: N815 + suit: SuitDict + name: str + slots: Dict[str, OdysseyWeapon] + + +class SuitDict(TypedDict): + """Dict representing a single suit.""" + + name: str + locName: str # noqa: N815 + edmcName: str # noqa: N815 + id: Any # ??? some sort of ID, not listed as to where or what + suitId: int # noqa: N815 + mods: List[str] + + +_OdysseyWeaponClassField = TypedDict('_OdysseyWeaponClassField', {'class': int}) + + +class OdysseyWeapon(_OdysseyWeaponClassField): + """Suit Weapon for an odyssey suit loadout.""" + + name: str + locName: str # noqa: N815 + id: Any + weaponrackId: int # noqa: N815 + locDescription: str # noqa: N815 + # class: int Oh this'll be fun. -- See the definition of the TypedDict this inherits from + mods: List[str] + + +class _ModuleEngineeringModifiers(TypedDict): + """Engineering modifiers for (ship) modules.""" + Label: str + Value: float + OriginalValue: float + LessIsGood: int + + +class _ModuleExperimentalEffects(TypedDict, total=False): + """Experimental effects an engineered (ship) module *MAY* have.""" + + ExperimentalEffect: str + ExperimentalEffect_Localised: str + + +class ModuleEngineering(_ModuleExperimentalEffects): + """Engineering modifiers for a module.""" + + Engineer: str + EngineerID: int + BlueprintName: str + BlueprintID: int + Level: int + Quality: int + Modifiers: List[_ModuleEngineeringModifiers] + + +class _ModulesOptionals(TypedDict, total=False): + """Optional fields a (ship) module may have.""" + + On: bool + Priority: int + Health: float + Value: int + Engineering: ModuleEngineering + + +class _ModulesWeaponsOptionals(TypedDict, total=False): + """Optional fields (ship) modules *may* have if they are weapons.""" + + AmmoInClip: int + AmmoInHopper: int + + +class ModuleDict(_ModulesOptionals, _ModulesWeaponsOptionals): + """Dictionary containing module information.""" + + Item: str + Slot: str diff --git a/plugins/eddb.py b/plugins/eddb.py index 80fd943151..c8792589c1 100644 --- a/plugins/eddb.py +++ b/plugins/eddb.py @@ -33,6 +33,7 @@ import plug from companion import CAPIData from config import config +from monitor_state_dict import MonitorStateDict if TYPE_CHECKING: from tkinter import Tk @@ -93,7 +94,7 @@ def prefs_changed(cmdr, is_beta): pass -def journal_entry(cmdr, is_beta, system, station, entry, state): +def journal_entry(cmdr, is_beta, system, station, entry, state: MonitorStateDict): # noqa: CCR001 D103 if (ks := killswitch.get_disabled('plugins.eddb.journal')).disabled: logger.warning(f'Journal processing for EDDB has been disabled: {ks.reason}') # LANG: Journal Processing disabled due to an active killswitch diff --git a/plugins/eddn.py b/plugins/eddn.py index 194a52deb3..f279b3c840 100644 --- a/plugins/eddn.py +++ b/plugins/eddn.py @@ -24,6 +24,7 @@ from config import applongname, appversion_nobuild, config, debug_senders from EDMCLogging import get_main_logger from monitor import monitor +from monitor_state_dict import MonitorStateDict from myNotebook import Frame from prefs import prefsVersion from ttkHyperlinkLabel import HyperlinkLabel @@ -773,7 +774,7 @@ def journal_entry( # noqa: C901, CCR001 system: str, station: str, entry: MutableMapping[str, Any], - state: Mapping[str, Any] + state: MonitorStateDict ) -> Optional[str]: """ Process a new Journal entry. diff --git a/plugins/edsm.py b/plugins/edsm.py index 30340925a1..3c15856aea 100644 --- a/plugins/edsm.py +++ b/plugins/edsm.py @@ -26,6 +26,7 @@ from config import applongname, appversion, config, debug_senders, trace_on from edmc_data import DEBUG_WEBSERVER_HOST, DEBUG_WEBSERVER_PORT from EDMCLogging import get_main_logger +from monitor_state_dict import MonitorStateDict from ttkHyperlinkLabel import HyperlinkLabel if TYPE_CHECKING: @@ -393,7 +394,7 @@ def credentials(cmdr: str) -> Optional[Tuple[str, str]]: def journal_entry( # noqa: C901, CCR001 - cmdr: str, is_beta: bool, system: str, station: str, entry: MutableMapping[str, Any], state: Mapping[str, Any] + cmdr: str, is_beta: bool, system: str, station: str, entry: MutableMapping[str, Any], state: MonitorStateDict ) -> None: """Journal Entry hook.""" if (ks := killswitch.get_disabled('plugins.edsm.journal')).disabled: diff --git a/plugins/inara.py b/plugins/inara.py index ea46f9ce48..ead8431511 100644 --- a/plugins/inara.py +++ b/plugins/inara.py @@ -21,6 +21,7 @@ from companion import CAPIData from config import applongname, appversion, config, debug_senders from EDMCLogging import get_main_logger +from monitor_state_dict import MonitorStateDict from ttkHyperlinkLabel import HyperlinkLabel logger = get_main_logger() @@ -327,7 +328,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: MonitorStateDict ) -> str: """ Journal entry hook. @@ -443,7 +444,7 @@ def journal_entry( # noqa: C901, CCR001 if state['Engineers']: # Not populated < 3.3 to_send_list: List[Mapping[str, Any]] = [] for k, v in state['Engineers'].items(): - e = {'engineerName': k} + e: Dict[str, Any] = {'engineerName': k} if isinstance(v, tuple): e['rankValue'] = v[0] @@ -726,9 +727,9 @@ def journal_entry( # noqa: C901, CCR001 new_add_event('setCommanderInventoryMaterials', entry['timestamp'], materials) this.materials = materials - except Exception as e: - logger.debug('Adding events', exc_info=e) - return str(e) + except Exception as ex: + logger.debug('Adding events', exc_info=ex) + return str(ex) # Send credits and stats to Inara on startup only - otherwise may be out of date if event_name == 'LoadGame': @@ -1113,7 +1114,7 @@ def journal_entry( # noqa: C901, CCR001 if not all(t in entry for t in ('Components', 'Consumables', 'Data', 'Items')): # So it's an empty event, core EDMC should have stuffed the data # into state['ShipLockerJSON']. - entry = state['ShipLockerJSON'] + entry = dict(state['ShipLockerJSON']) odyssey_plural_microresource_types = ('Items', 'Components', 'Data', 'Consumables') # we're getting new data here. so reset it on inara's side just to be sure that we set everything right @@ -1361,7 +1362,7 @@ def cmdr_data(data: CAPIData, is_beta): # noqa: CCR001 this.lastcredits = int(data['commander']['credits']) -def make_loadout(state: Dict[str, Any]) -> OrderedDictT[str, Any]: # noqa: CCR001 +def make_loadout(state: Mapping[str, Any]) -> OrderedDictT[str, Any]: # noqa: CCR001 """ Construct an inara loadout from an event.