#!/usr/bin/env python3 """ EDMC.py - Command-line interface. Requires prior setup through the GUI. Copyright (c) EDCD, All Rights Reserved Licensed under the GNU General Public License v2 or later. See LICENSE file. """ from __future__ import annotations import argparse import json import os import queue import sys import warnings from enum import IntEnum from pathlib import Path from time import sleep from datetime import datetime, timezone from typing import TYPE_CHECKING, Any from common_utils import log_locale, SERVER_RETRY from contextlib import suppress # isort: off os.environ["EDMC_NO_UI"] = "1" # See EDMCLogging.py docs. # workaround for https://github.com/EDCD/EDMarketConnector/issues/568 from EDMCLogging import edmclogger, logger, logging if TYPE_CHECKING: from logging import TRACE # type: ignore # noqa: F401 # needed to make mypy happy edmclogger.set_channels_loglevel(logging.INFO) # isort: on import collate import commodity import companion import edshipyard from l10n import translations as tr import loadout import outfitting import shipyard import stats from config import appcmdname, appversion, config from monitor import monitor from update import EDMCVersion, Updater, check_for_fdev_updates, check_for_datafile_updates sys.path.append(config.internal_plugin_dir) # This import must be after the sys.path.append. # The sys.path.append has to be after `import sys` and `from config import config` # isort: off import eddn # noqa: E402 # isort: on tr.install_dummy() class ExitCode(IntEnum): """List the various available exit codes.""" SUCCESS = 0 SERVER = 1 CREDENTIALS = 2 VERIFICATION = 3 LAGGING = 4 SYS_ERR = 5 ARGS = 6 JOURNAL_READ_ERR = 7 COMMANDER_UNKNOWN = 8 # Backwards Compatibility EXIT_SUCCESS = ExitCode.SUCCESS EXIT_SERVER = ExitCode.SERVER EXIT_CREDENTIALS = ExitCode.CREDENTIALS EXIT_VERIFICATION = ExitCode.VERIFICATION EXIT_LAGGING = ExitCode.LAGGING EXIT_SYS_ERR = ExitCode.SYS_ERR EXIT_ARGS = ExitCode.ARGS EXIT_JOURNAL_READ_ERR = ExitCode.JOURNAL_READ_ERR EXIT_COMMANDER_UNKNOWN = ExitCode.COMMANDER_UNKNOWN def deep_get(target: companion.CAPIData, *args: str, default=None) -> Any: """ Walk into a dict and return the specified deep value. Deprecated! Will be removed in 6.4 or later. Use structural pattern matching or explicit dict `.get()` chains instead. """ warnings.warn( "deep_get is deprecated and will be removed in a future version. " "Use modern pattern matching (match/case) or sequential dict.get() chains instead.", category=DeprecationWarning, stacklevel=2 ) if not hasattr(target, 'get'): raise ValueError(f"Cannot call get on {target} ({type(target)})") current = target for arg in args: match current: case dict() | companion.CAPIData(): if (res := current.get(arg)) is None: return default current = res case _: return default return current def parse_cli_args() -> argparse.Namespace: """Parse and return command-line arguments.""" parser = argparse.ArgumentParser( prog=appcmdname, description='Prints the current system and station (if docked) to stdout and optionally writes player ' 'status, ship locations, ship loadout and/or station data to file. ' 'Requires prior setup through the accompanying GUI app.' ) parser.add_argument('-v', '--version', help='print program version and exit', action='store_const', const=True) group_loglevel = parser.add_mutually_exclusive_group() group_loglevel.add_argument( '--loglevel', metavar='loglevel', help='Set the logging loglevel to one of: CRITICAL, ERROR, WARNING, INFO, DEBUG, TRACE' ) parser.add_argument('--trace', help='Set the Debug logging loglevel to TRACE', action='store_true') parser.add_argument('--trace-on', action='append', help='Mark the selected trace logging as active. "*" or "all" is equivalent to --trace-all') parser.add_argument("--trace-all", action='store_true', help='Force trace level logging, with all possible --trace-on values active.') parser.add_argument("--trace-help", action='store_true', help="Open the documentation to view all possible trace-on values") parser.add_argument('--refresh-all', action='store_true', help='Iterate over all known CMDRs to try and refresh access tokens') parser.add_argument('--config', help="Define a custom config to load EDMC from") parser.add_argument('-a', metavar='FILE', help='write ship loadout to FILE in Companion API json format') parser.add_argument('-e', metavar='FILE', help='write ship loadout to FILE in E:D Shipyard plain text format') parser.add_argument('-l', metavar='FILE', help='write ship locations to FILE in CSV format') parser.add_argument('-m', metavar='FILE', help='write station commodity market data to FILE in CSV format') parser.add_argument('-o', metavar='FILE', help='write station outfitting data to FILE in CSV format') parser.add_argument('-s', metavar='FILE', help='write station shipyard data to FILE in CSV format') parser.add_argument('-t', metavar='FILE', help='write player status to FILE in CSV format') parser.add_argument('-d', metavar='FILE', help='write raw JSON data to FILE') parser.add_argument('-n', action='store_true', help='send data to EDDN') parser.add_argument('-p', metavar='CMDR', help='Returns data from the specified player account') parser.add_argument('-j', help=argparse.SUPPRESS) # Import JSON dump return parser.parse_args() def handle_version_check() -> None: """Check for appcast updates and print version strings.""" updater = Updater() newversion: EDMCVersion | None = updater.check_appcast() if newversion: # LANG: Update Available Text newverstr: str = tr.tl("{NEWVER} is available").format(NEWVER=newversion.title) print(f'{appversion()} ({newverstr})') else: print(appversion()) def _determine_log_level(args: argparse.Namespace) -> int | str | None: """Evaluate the correct log level with clean early-exit precedence.""" if args.trace_all or (args.trace_on and any(x in args.trace_on for x in ('*', 'all'))): logger.info('Setting TRACE_ALL level debugging due to --trace-all or wildcard flag') return logging.TRACE_ALL # type: ignore if args.trace or args.trace_on: logger.info('Setting TRACE level debugging due to --trace or --trace-on') return logging.TRACE # type: ignore return args.loglevel def configure_logging_and_env(args: argparse.Namespace) -> None: """Configure logger log levels, trace environments, and initial config payloads.""" if level := _determine_log_level(args): if isinstance(level, int): logger.setLevel(level) edmclogger.set_channels_loglevel(level) else: edmclogger.set_channels_loglevel(level) # Environment Configuration updates if args.config: config.reload_from_path(args.config) logger.debug(f'Startup v{appversion()} : Running on Python v{sys.version}') logger.debug( f'Platform: {sys.platform}\n' f'argv[0]: {sys.argv[0]}\n' f'exec_prefix: {sys.exec_prefix}\n' f'executable: {sys.executable}\n' f'sys.path: {sys.path}' ) if args.trace_on: import config as conf_module conf_module.trace_on = [x.casefold() for x in args.trace_on] for trace_item in conf_module.trace_on: logger.info(f'marked {trace_item} for TRACE') log_locale('Initial Locale') if args.trace_help: import webbrowser logger.info("Opening Trace Help Documentation") webbrowser.open("https://github.com/EDCD/EDMarketConnector/blob/main/docs/Available%20Traces.md") def refresh_all_cmdrs() -> None: """Attempt to iterate and refresh access tokens for all known commanders.""" logger.debug("Refreshing all known CMDRs") cmdrs = config.get_list('cmdrs', default=[]) for cmdr in cmdrs: logger.debug(f'Attempting to use commander "{cmdr}"') try: companion.session.login(cmdr, monitor.is_beta) logger.debug("Succeeded!") except AttributeError: logger.debug(f"Unable to refresh CMDR {cmdr}.") def load_json_dump(json_file_path: str) -> companion.CAPIData: """Load and return CAPI data from a local JSON dump file.""" logger.debug('Import and collate from JSON dump') json_file = str(Path(json_file_path).resolve()) data = None with suppress(UnicodeDecodeError), open(json_file, encoding='utf-8') as file_handle: data = json.load(file_handle) if data is None: with open(json_file, encoding='utf-8') as file_handle: data = json.load(file_handle) config.set('querytime', int(Path(json_file_path).stat().st_mtime)) return data def load_state_from_journal_and_capi(args: argparse.Namespace) -> companion.CAPIData: """Read the latest Journal state and fetch live CAPI data.""" logger.debug('Getting state from latest journal file') try: monitor.currentdir = config.get_str('journaldir', default=config.default_journal_dir) or config.default_journal_dir logger.debug(f'logdir = "{monitor.currentdir}"') if (logfile := monitor.journal_newest_filename(monitor.currentdir)) is None: raise ValueError("None from monitor.journal_newest_filename") logger.debug(f'Using logfile "{logfile}"') with open(logfile, 'rb', 0) as loghandle: for line in loghandle: with suppress(Exception): monitor.parse_entry(line) continue logger.debug(f'Invalid journal entry {line!r}') except Exception: logger.exception("Can't read Journal file") sys.exit(ExitCode.JOURNAL_READ_ERR) if not monitor.cmdr: logger.error('Not available while E:D is at the main menu') sys.exit(ExitCode.COMMANDER_UNKNOWN) cmdrs = config.get_list('cmdrs', default=[]) target_cmdr = args.p or monitor.cmdr logger.debug(f'Attempting to use commander "{target_cmdr}"' + (' from Journal File' if not args.p else '')) try: cmdr_to_use = next(c for c in cmdrs if c.lower() == target_cmdr.lower()) except StopIteration: raise companion.CredentialsError() from StopIteration try: companion.session.login(cmdr_to_use, monitor.is_beta) except AttributeError: raise companion.CredentialsError() from AttributeError # Initiate CAPI queries querytime = int(datetime.now(timezone.utc).timestamp()) companion.session.station(query_time=querytime) # Wait for the response _capi_request_timeout = 60 try: capi_response = companion.session.capi_response_queue.get(block=True, timeout=_capi_request_timeout) except queue.Empty: logger.error(f'CAPI requests timed out after {_capi_request_timeout} seconds') sys.exit(ExitCode.SERVER) match capi_response: case companion.EDMCCAPIFailedRequest(message=msg, exception=exc): logger.trace_if('capi.worker', f'Failed Request: {msg}') if exc: raise exc raise ValueError(msg) case companion.EDMCCAPIResponse(capi_data=capi_data): logger.trace_if('capi.worker', 'Answer is not a Failure') config.set('querytime', querytime) return capi_data case _: raise ValueError(f"Response was neither CAPIFailedRequest nor EDMCAPIResponse: {type(capi_response)}") def validate_capi_data(data: companion.CAPIData, is_json_import: bool) -> None: """Validate structural integrity and data lagging utilizing pattern matching.""" # Check Commander Name match data: case {"commander": {"name": str(name)}} if name.strip(): pass case _: logger.error("No data['commander']['name'] from CAPI") sys.exit(ExitCode.SERVER) # Check System match data: case {"lastSystem": {"name": str(sys_name)}} if sys_name.strip(): pass case _: logger.error("No data['lastSystem']['name'] from CAPI") sys.exit(ExitCode.SERVER) # Check Docking & Starport match data: case {"commander": {"docked": True}}: match data: case {"lastStarport": {"name": str(port_name)}} if port_name.strip(): pass case _: logger.error("No data['lastStarport']['name'] from CAPI (Required when docked)") sys.exit(ExitCode.SERVER) case _: pass # Check Ship Loadout Structure match data: case {"ship": {"modules": dict() | list(), "name": str(ship_name)}} if ship_name.strip(): pass case _: logger.error("No data['ship']['modules'] or data['ship']['name'] from CAPI") sys.exit(ExitCode.SERVER) # If importing a JSON dump, bypass lagging/journal validations if is_json_import: return # Match Commander exactly to Journal if data["commander"]["name"] != monitor.cmdr: raise companion.CmdrError() # Determine if server is lagging behind the journal is_docked = data["commander"].get("docked", False) capi_station = data.get("lastStarport", {}).get("name") if is_docked else None if ( data["lastSystem"]["name"] != monitor.state['SystemName'] or capi_station != monitor.state['StationName'] or data["ship"]["id"] != monitor.state['ShipID'] or data["ship"]["name"].lower() != monitor.state['ShipType'] ): raise companion.ServerLagging() def export_standard_data(data: companion.CAPIData, args: argparse.Namespace) -> None: """Handle all standard, non-docked specific file export operations.""" if args.d: logger.debug(f'Writing raw JSON data to "{args.d}"') out = json.dumps(dict(data), ensure_ascii=False, indent=2, sort_keys=True, separators=(',', ': ')) with open(args.d, 'wb') as f: f.write(out.encode("utf-8")) if args.a: logger.debug(f'Writing Ship Loadout in Companion API JSON format to "{args.a}"') loadout.export(data, args.a) if args.e: logger.debug(f'Writing Ship Loadout in ED Shipyard plain text format to "{args.e}"') edshipyard.export(data, args.e) if args.l: logger.debug(f'Writing Ship Locations in CSV format to "{args.l}"') stats.export_ships(data, args.l) if args.t: logger.debug(f'Writing Player Status in CSV format to "{args.t}"') stats.export_status(data, args.t) # Console stdout behavior based on docked state match data: case {"commander": {"docked": True}, "lastSystem": {"name": sys_name}, "lastStarport": {"name": port_name}}: print(f'{sys_name},{port_name}') case {"lastSystem": {"name": sys_name}}: print(f'{sys_name}') case _: print('Unknown') def export_docked_data(data: companion.CAPIData, args: argparse.Namespace) -> None: # noqa: C901, CCR001 """Process market, outfitting, shipyard and EDDN exports requiring a docked state.""" # Ensure they've requested one of the docked flags if not (args.m or args.o or args.s or args.n or args.j): return match data: case {"commander": {"docked": True}, "lastStarport": {"name": port_name}} if port_name.strip(): pass # Valid Docking Base case {"commander": {"docked": False}}: logger.error("Can't use -m, -o, -s, -n or -j because you're not currently docked!") return case _: logger.error("No data['lastStarport']['name'] from CAPI") sys.exit(ExitCode.LAGGING) # Ensure commodity or outfitting data is present in the CAPI payload has_commodities = bool(data["lastStarport"].get("commodities")) has_modules = bool(data["lastStarport"].get("modules")) if not (has_commodities or has_modules): logger.error("No commodities or outfitting (modules) in CAPI data") return if args.j: logger.debug('Importing data from the CAPI return...') collate.addcommodities(data) collate.addmodules(data) collate.addships(data) if args.m: logger.debug(f'Writing Station Commodity Market Data in CSV format to "{args.m}"') if has_commodities: fixed = companion.fixup(data) mkt_type = config.get_str('mkt_export_type', default='CSV') try: kind = commodity.CommodityExportKind[mkt_type] except KeyError: kind = commodity.CommodityExportKind.SEMICOLON commodity.export(fixed, kind, args.m) else: logger.error("Station doesn't have a market") if args.o: if has_modules: logger.debug(f'Writing Station Outfitting in CSV format to "{args.o}"') outfitting.export(data, args.o) # type: ignore else: logger.error("Station doesn't supply outfitting") # Retry logic for Shipyard if missing but reported as a station service has_ships = bool(data["lastStarport"].get("ships")) has_shipyard_service = bool(data["lastStarport"].get("services", {}).get("shipyard")) if (args.s or args.n) and not args.j and not has_ships and has_shipyard_service: sleep(SERVER_RETRY) companion.session.station(int(datetime.now(timezone.utc).timestamp())) try: capi_response = companion.session.capi_response_queue.get(block=True, timeout=60) except queue.Empty: logger.error('CAPI requests timed out after 60 seconds (Shipyard Retry)') sys.exit(ExitCode.SERVER) if isinstance(capi_response, companion.EDMCCAPIFailedRequest): logger.error(f'Failed Request: {capi_response.message}') sys.exit(ExitCode.SERVER) new_data = capi_response.capi_data # Verify the player didn't undock or jump during the sleep cycle match new_data: case { "commander": {"docked": True}, "lastSystem": {"name": sys_name}, "lastStarport": {"name": port_name} } if sys_name == monitor.state['SystemName'] and port_name == monitor.state['StationName']: data = new_data if args.s: match data: case {"lastStarport": {"ships": {"shipyard_list": shipyard_list}}} if shipyard_list: logger.debug(f'Writing Station Shipyard in CSV format to "{args.s}"') shipyard.export(data, args.s) case _: if not args.j and monitor.stationservices and 'Shipyard' in monitor.stationservices: logger.error('Failed to get shipyard data') else: logger.error("Station doesn't have a shipyard") if args.n: try: eddn_sender = eddn.EDDN(None) logger.debug('Sending Market, Outfitting and Shipyard data to EDDN...') eddn_sender.export_commodities(data, monitor.is_beta) eddn_sender.export_outfitting(data, monitor.is_beta) eddn_sender.export_shipyard(data, monitor.is_beta) except Exception: logger.exception('Failed to send data to EDDN') def main() -> None: """Run the main code of the program.""" try: args = parse_cli_args() if args.version: handle_version_check() return configure_logging_and_env(args) if args.trace_help: return # Help browser launched, terminate sequence if args.refresh_all: refresh_all_cmdrs() if args.j: data = load_json_dump(args.j) else: data = load_state_from_journal_and_capi(args) validate_capi_data(data, is_json_import=bool(args.j)) export_standard_data(data, args) export_docked_data(data, args) except companion.ServerConnectionError: logger.exception('Exception while contacting server') sys.exit(ExitCode.SERVER) except companion.ServerError: logger.exception('Frontier CAPI Server returned an error') sys.exit(ExitCode.SERVER) except companion.CredentialsError: logger.error('Frontier CAPI Server: Invalid Credentials') sys.exit(ExitCode.CREDENTIALS) except companion.ServerLagging: logger.error( 'Mismatch(es) between CAPI and Journal for at least one of: ' 'StarSystem, Last Star Port, Ship ID or Ship Name/Type' ) sys.exit(ExitCode.SERVER) except companion.CmdrError: logger.error( f'Commander "{data["commander"]["name"]}" from CAPI doesn\'t match ' f'"{monitor.cmdr}" from Journal' ) sys.exit(ExitCode.SERVER) except Exception: logger.exception('"other" exception') sys.exit(ExitCode.SERVER) if __name__ == '__main__': try: check_for_fdev_updates(silent=True) check_for_datafile_updates(silent=True) main() except KeyboardInterrupt: logger.info("Ctrl+C Detected, Attempting Clean Shutdown") logger.debug('Exiting') sys.exit(ExitCode.SUCCESS)