EDCD#2051 First Pass Penultimate

EDMCLogging.py

@@ -34,6 +34,7 @@
     # See, plug.py:load_plugins()
     logger = logging.getLogger(f'{appname}.{plugin_name}')
 """
+from __future__ import annotations
 
 import inspect
 import logging
@@ -183,18 +184,12 @@ def __init__(self, logger_name: str, loglevel: int | str = _default_loglevel):
         # rotated versions.
         # This is {logger_name} so that EDMC.py logs to a different file.
         logfile_rotating = pathlib.Path(tempfile.gettempdir())
-        logfile_rotating = logfile_rotating / f'{appname}'
+        logfile_rotating /= f'{appname}'
         logfile_rotating.mkdir(exist_ok=True)
-        logfile_rotating = logfile_rotating / f'{logger_name}-debug.log'
-
-        self.logger_channel_rotating = logging.handlers.RotatingFileHandler(
-            logfile_rotating,
-            mode='a',
-            maxBytes=1024 * 1024,  # 1MiB
-            backupCount=10,
-            encoding='utf-8',
-            delay=False
-        )
+        logfile_rotating /= f'{logger_name}-debug.log'
+
+        self.logger_channel_rotating = logging.handlers.RotatingFileHandler(logfile_rotating, maxBytes=1024 * 1024,
+                                                                            backupCount=10, encoding='utf-8')
         # Yes, we always want these rotated files to be at TRACE level
         self.logger_channel_rotating.setLevel(logging.TRACE)  # type: ignore
         self.logger_channel_rotating.setFormatter(self.logger_formatter)
@@ -535,9 +530,8 @@ def get_main_logger(sublogger_name: str = '') -> 'LoggerMixin':
     if not os.getenv("EDMC_NO_UI"):
         # GUI app being run
         return cast('LoggerMixin', logging.getLogger(appname))
-    else:
-        # Must be the CLI
-        return cast('LoggerMixin', logging.getLogger(appcmdname))
+    # Must be the CLI
+    return cast('LoggerMixin', logging.getLogger(appcmdname))
 
 
 # Singleton

collate.py

@@ -79,7 +79,7 @@ def addcommodities(data) -> None:  # noqa: CCR001
 
         commodities[key] = new
 
-    if not len(commodities) > size_pre:
+    if len(commodities) <= size_pre:
         return
 
     if isfile(commodityfile):
@@ -227,7 +227,7 @@ def addships(data) -> None:  # noqa: CCR001
             print('Not docked!')
             continue
 
-        elif not data.get('lastStarport'):
+        if not data.get('lastStarport'):
             print('No starport!')
             continue
 

companion.py

@@ -5,6 +5,7 @@
 Some associated code is in protocol.py which creates and handles the edmc://
 protocol used for the callback.
 """
+from __future__ import annotations
 
 import base64
 import collections
@@ -44,7 +45,6 @@ def _(x): return x
 else:
     UserDict = collections.UserDict  # type: ignore # Otherwise simply use the actual class
 
-
 capi_query_cooldown = 60  # Minimum time between (sets of) CAPI queries
 capi_fleetcarrier_query_cooldown = 60 * 15  # Minimum time between CAPI fleetcarrier queries
 capi_default_requests_timeout = 10
@@ -195,10 +195,10 @@ def listify(thing: Union[List, Dict]) -> List:
     if thing is None:
         return []  # data is not present
 
-    elif isinstance(thing, list):
+    if isinstance(thing, list):
         return list(thing)  # array is not sparse
 
-    elif isinstance(thing, dict):
+    if isinstance(thing, dict):
         retval: List[Any] = []
         for k, v in thing.items():
             idx = int(k)
@@ -211,8 +211,7 @@ def listify(thing: Union[List, Dict]) -> List:
 
         return retval
 
-    else:
-        raise ValueError(f"expected an array or sparse array, got {thing!r}")
+    raise ValueError(f"expected an array or sparse array, got {thing!r}")
 
 
 class ServerError(Exception):
@@ -346,7 +345,7 @@ def refresh(self) -> Optional[str]:
         logger.debug(f'idx = {idx}')
 
         tokens = config.get_list('fdev_apikeys', default=[])
-        tokens = tokens + [''] * (len(cmdrs) - len(tokens))
+        tokens += [''] * (len(cmdrs) - len(tokens))
         if tokens[idx]:
             logger.debug('We have a refresh token for that idx')
             data = {
@@ -371,9 +370,8 @@ def refresh(self) -> Optional[str]:
 
                     return data.get('access_token')
 
-                else:
-                    logger.error(f"Frontier CAPI Auth: Can't refresh token for \"{self.cmdr}\"")
-                    self.dump(r)
+                logger.error(f"Frontier CAPI Auth: Can't refresh token for \"{self.cmdr}\"")
+                self.dump(r)
 
             except (ValueError, requests.RequestException, ) as e:
                 logger.exception(f"Frontier CAPI Auth: Can't refresh token for \"{self.cmdr}\"\n{e!r}")
@@ -489,7 +487,7 @@ def authorize(self, payload: str) -> str:  # noqa: CCR001
                 cmdrs = config.get_list('cmdrs', default=[])
                 idx = cmdrs.index(self.cmdr)
                 tokens = config.get_list('fdev_apikeys', default=[])
-                tokens = tokens + [''] * (len(cmdrs) - len(tokens))
+                tokens += [''] * (len(cmdrs) - len(tokens))
                 tokens[idx] = data_token.get('refresh_token', '')
                 config.set('fdev_apikeys', tokens)
                 config.save()  # Save settings now for use by command-line app
@@ -530,7 +528,7 @@ def invalidate(cmdr: Optional[str]) -> None:
             cmdrs = config.get_list('cmdrs', default=[])
             idx = cmdrs.index(cmdr)
             to_set = config.get_list('fdev_apikeys', default=[])
-            to_set = to_set + [''] * (len(cmdrs) - len(to_set))  # type: ignore
+            to_set += [''] * (len(cmdrs) - len(to_set))  # type: ignore
             to_set[idx] = ''
 
         if to_set is None:
@@ -693,7 +691,7 @@ def login(self, cmdr: Optional[str] = None, is_beta: Optional[bool] = None) -> b
                 logger.error('self.credentials is None')
                 raise CredentialsError('Missing credentials')  # Shouldn't happen
 
-            elif self.state == Session.STATE_OK:
+            if self.state == Session.STATE_OK:
                 logger.debug('already logged in (state == STATE_OK)')
                 return True  # already logged in
 
@@ -703,10 +701,9 @@ def login(self, cmdr: Optional[str] = None, is_beta: Optional[bool] = None) -> b
                 logger.debug(f'already logged in (is_beta = {is_beta})')
                 return True  # already logged in
 
-            else:
-                logger.debug('changed account or retrying login during auth')
-                self.reinit_session()
-                self.credentials = credentials
+            logger.debug('changed account or retrying login during auth')
+            self.reinit_session()
+            self.credentials = credentials
 
         self.state = Session.STATE_INIT
         self.auth = Auth(self.credentials['cmdr'])
@@ -718,11 +715,10 @@ def login(self, cmdr: Optional[str] = None, is_beta: Optional[bool] = None) -> b
             self.start_frontier_auth(access_token)
             return True
 
-        else:
-            logger.debug('We do NOT have an access_token')
-            self.state = Session.STATE_AUTH
-            return False
-            # Wait for callback
+        logger.debug('We do NOT have an access_token')
+        self.state = Session.STATE_AUTH
+        return False
+        # Wait for callback
 
     # Callback from protocol handler
     def auth_callback(self) -> None:
@@ -934,9 +930,8 @@ def capi_station_queries(  # noqa: CCR001
                     logger.warning(f"{last_starport_id!r} != {int(market_data['id'])!r}")
                     raise ServerLagging()
 
-                else:
-                    market_data['name'] = last_starport_name
-                    station_data['lastStarport'].update(market_data)
+                market_data['name'] = last_starport_name
+                station_data['lastStarport'].update(market_data)
 
             if services.get('outfitting') or services.get('shipyard'):
                 shipyard_data = capi_single_query(capi_host, self.FRONTIER_CAPI_PATH_SHIPYARD, timeout=timeout)
@@ -948,9 +943,8 @@ def capi_station_queries(  # noqa: CCR001
                     logger.warning(f"{last_starport_id!r} != {int(shipyard_data['id'])!r}")
                     raise ServerLagging()
 
-                else:
-                    shipyard_data['name'] = last_starport_name
-                    station_data['lastStarport'].update(shipyard_data)
+                shipyard_data['name'] = last_starport_name
+                station_data['lastStarport'].update(shipyard_data)
             # WORKAROUND END
 
             return station_data
@@ -1173,11 +1167,9 @@ def capi_host_for_galaxy(self) -> str:
             logger.debug(f"Using {SERVER_LIVE} because monitor.is_live_galaxy() was True")
             return SERVER_LIVE
 
-        else:
-            logger.debug(f"Using {SERVER_LEGACY} because monitor.is_live_galaxy() was False")
-            return SERVER_LEGACY
+        logger.debug(f"Using {SERVER_LEGACY} because monitor.is_live_galaxy() was False")
+        return SERVER_LEGACY
 
-        return ''
     ######################################################################
 
 
@@ -1194,7 +1186,7 @@ def fixup(data: CAPIData) -> CAPIData:  # noqa: C901, CCR001 # Can't be usefully
     if not commodity_map:
         # Lazily populate
         for f in ('commodity.csv', 'rare_commodity.csv'):
-            with open(config.respath_path / 'FDevIDs' / f, 'r') as csvfile:
+            with open(config.respath_path / 'FDevIDs' / f) as csvfile:
                 reader = csv.DictReader(csvfile)
 
                 for row in reader:
@@ -1315,11 +1307,10 @@ def index_possibly_sparse_list(data: Union[Mapping[str, V], List[V]], key: int)
     if isinstance(data, list):
         return data[key]
 
-    elif isinstance(data, (dict, OrderedDict)):
+    if isinstance(data, (dict, OrderedDict)):
         return data[str(key)]
 
-    else:
-        raise ValueError(f'Unexpected data type {type(data)}')
+    raise ValueError(f'Unexpected data type {type(data)}')
 ######################################################################
 
 

config/__init__.py

@@ -6,6 +6,7 @@
 macOS uses a 'defaults' object.
 """
 
+from __future__ import annotations
 
 __all__ = [
     # defined in the order they appear in the file
@@ -90,13 +91,10 @@ def git_shorthash_from_head() -> str:
     shorthash: str = None  # type: ignore
 
     try:
-        git_cmd = subprocess.Popen('git rev-parse --short HEAD'.split(),
-                                   stdout=subprocess.PIPE,
-                                   stderr=subprocess.STDOUT
-                                   )
-        out, err = git_cmd.communicate()
+        result = subprocess.run(['git', 'rev-parse', '--short', 'HEAD'], capture_output=True, text=True)
+        out = result.stdout.strip()
 
-    except Exception as e:
+    except subprocess.CalledProcessError as e:
         logger.info(f"Couldn't run git command for short hash: {e!r}")
 
     else:
@@ -325,13 +323,13 @@ def get(
         if (a_list := self._suppress_call(self.get_list, ValueError, key, default=None)) is not None:
             return a_list
 
-        elif (a_str := self._suppress_call(self.get_str, ValueError, key, default=None)) is not None:
+        if (a_str := self._suppress_call(self.get_str, ValueError, key, default=None)) is not None:
             return a_str
 
-        elif (a_bool := self._suppress_call(self.get_bool, ValueError, key, default=None)) is not None:
+        if (a_bool := self._suppress_call(self.get_bool, ValueError, key, default=None)) is not None:
             return a_bool
 
-        elif (an_int := self._suppress_call(self.get_int, ValueError, key, default=None)) is not None:
+        if (an_int := self._suppress_call(self.get_int, ValueError, key, default=None)) is not None:
             return an_int
 
         return default  # type: ignore
@@ -461,16 +459,15 @@ def get_config(*args, **kwargs) -> AbstractConfig:
         from .darwin import MacConfig
         return MacConfig(*args, **kwargs)
 
-    elif sys.platform == "win32":  # pragma: sys-platform-win32
+    if sys.platform == "win32":  # pragma: sys-platform-win32
         from .windows import WinConfig
         return WinConfig(*args, **kwargs)
 
-    elif sys.platform == "linux":  # pragma: sys-platform-linux
+    if sys.platform == "linux":  # pragma: sys-platform-linux
         from .linux import LinuxConfig
         return LinuxConfig(*args, **kwargs)
 
-    else:  # pragma: sys-platform-not-known
-        raise ValueError(f'Unknown platform: {sys.platform=}')
+    raise ValueError(f'Unknown platform: {sys.platform=}')
 
 
 config = get_config()

config/darwin.py

@@ -99,7 +99,7 @@ def get_list(self, key: str, *, default: list = None) -> list:
         if res is None:
             return default  # type: ignore # Yes it could be None, but we're _assuming_ that people gave us a default
 
-        elif not isinstance(res, list):
+        if not isinstance(res, list):
             raise ValueError(f'__raw_get returned unexpected type {type(res)=} {res!r}')
 
         return res
@@ -114,7 +114,7 @@ def get_int(self, key: str, *, default: int = 0) -> int:
         if res is None:
             return default
 
-        elif not isinstance(res, (str, int)):
+        if not isinstance(res, (str, int)):
             raise ValueError(f'__raw_get returned unexpected type {type(res)=} {res!r}')
 
         try:
@@ -134,7 +134,7 @@ def get_bool(self, key: str, *, default: bool = None) -> bool:
         if res is None:
             return default  # type: ignore # Yes it could be None, but we're _assuming_ that people gave us a default
 
-        elif not isinstance(res, bool):
+        if not isinstance(res, bool):
             raise ValueError(f'__raw_get returned unexpected type {type(res)=} {res!r}')
 
         return res

config/windows.py

@@ -1,4 +1,5 @@
 """Windows config implementation."""
+from __future__ import annotations
 
 # spell-checker: words folderid deps hkey edcd
 import ctypes
@@ -43,6 +44,7 @@ class WinConfig(AbstractConfig):
     """Implementation of AbstractConfig for Windows."""
 
     def __init__(self, do_winsparkle=True) -> None:
+        super().__init__()
         self.app_dir_path = pathlib.Path(str(known_folder_path(FOLDERID_LocalAppData))) / appname
         self.app_dir_path.mkdir(exist_ok=True)
 
@@ -132,15 +134,14 @@ def __get_regentry(self, key: str) -> Union[None, list, str, int]:
         if _type == winreg.REG_SZ:
             return str(value)
 
-        elif _type == winreg.REG_DWORD:
+        if _type == winreg.REG_DWORD:
             return int(value)
 
-        elif _type == winreg.REG_MULTI_SZ:
+        if _type == winreg.REG_MULTI_SZ:
             return list(value)
 
-        else:
-            logger.warning(f'registry key {key=} returned unknown type {_type=} {value=}')
-            return None
+        logger.warning(f'registry key {key=} returned unknown type {_type=} {value=}')
+        return None
 
     def get_str(self, key: str, *, default: str | None = None) -> str:
         """
@@ -152,7 +153,7 @@ def get_str(self, key: str, *, default: str | None = None) -> str:
         if res is None:
             return default  # type: ignore # Yes it could be None, but we're _assuming_ that people gave us a default
 
-        elif not isinstance(res, str):
+        if not isinstance(res, str):
             raise ValueError(f'Data from registry is not a string: {type(res)=} {res=}')
 
         return res
@@ -167,7 +168,7 @@ def get_list(self, key: str, *, default: list | None = None) -> list:
         if res is None:
             return default  # type: ignore # Yes it could be None, but we're _assuming_ that people gave us a default
 
-        elif not isinstance(res, list):
+        if not isinstance(res, list):
             raise ValueError(f'Data from registry is not a list: {type(res)=} {res}')
 
         return res