Merge pull request #277 from CompOmics/fix/ruff-errors

Resolve ruff linting errors

Author: RalfG

ms2rescore/__init__.py

@@ -27,8 +27,8 @@
     module="pyopenms",
 )
 
-from ms2rescore._version import get_version  # noqa: E402
-from ms2rescore.config_parser import parse_configurations  # noqa: E402
-from ms2rescore.core import rescore  # noqa: E402
+from ms2rescore._version import get_version
+from ms2rescore.config_parser import parse_configurations
+from ms2rescore.core import rescore
 
 __version__ = get_version()

ms2rescore/__main__.py

@@ -8,7 +8,6 @@
 import sys
 from datetime import datetime
 from pathlib import Path
-from typing import Union
 
 from rich.console import Console
 from rich.logging import RichHandler
@@ -23,7 +22,7 @@
 try:
     import matplotlib.pyplot as plt
 
-    plt.set_loglevel("warning")
+    plt.set_loglevel("WARNING")
 except ImportError:
     pass
 
@@ -164,7 +163,7 @@ def _argument_parser() -> argparse.ArgumentParser:
     return parser
 
 
-def _setup_logging(passed_level: str, log_file: Union[str, Path]):
+def _setup_logging(passed_level: str, log_file: str | Path):
     """Setup logging for writing to log file and Rich Console."""
     if passed_level not in LOG_MAPPING:
         raise MS2RescoreConfigurationError(
@@ -190,7 +189,7 @@ def inner(*args, **kwargs):
             return_value = fnc(*args, **kwargs)
 
         # Add timestamp to profiler output filename
-        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+        timestamp = datetime.now().astimezone().strftime("%Y%m%d_%H%M%S")
         profile_filename = f"{filepath}.profile_{timestamp}.prof"
         profiler.dump_stats(profile_filename)
         LOGGER.info(f"Profile data written to: {profile_filename}")
@@ -255,8 +254,8 @@ def main(tims=False):
             profiled_rescore(configuration=config)
         else:
             rescore(configuration=config)
-    except Exception as e:
-        LOGGER.exception(e)
+    except Exception:
+        LOGGER.exception("Unhandled error during rescoring")
         sys.exit(1)
     finally:
         CONSOLE.save_html(config["ms2rescore"]["output_path"] + ".log.html")

ms2rescore/_ristretto_utils.py

@@ -9,7 +9,6 @@
 """
 
 import logging
-from typing import Dict, Optional, Set
 
 import numpy as np
 import pandas as pd
@@ -25,7 +24,7 @@
 
 def _build_features_dataframe(
     psm_list: PSMList,
-    feature_names: Set[str],
+    feature_names: set[str],
     lower_score_is_better: bool,
 ) -> pd.DataFrame:
     """
@@ -71,9 +70,9 @@ def _trim_and_evaluate(
     max_rank: int,
     *,
     run_col: str,
-    peptide_col: Optional[str],
-    protein_col: Optional[str],
-    decoy_pattern: Optional[str],
+    peptide_col: str | None,
+    protein_col: str | None,
+    decoy_pattern: str | None,
 ) -> RescoreResult:
     """
     Compete to at most ``max_rank`` PSMs per spectrum, then compute q-values/PEP/rollups.
@@ -128,7 +127,7 @@ def _is_original_psm(psm) -> bool:
     return bool(value)
 
 
-def evaluate_before(psm_list: PSMList, config: Dict) -> RescoreResult:
+def evaluate_before(psm_list: PSMList, config: dict) -> RescoreResult:
     """
     Evaluate the PSMs' current (pre-rescoring) score with ristretto, for report baselines.
 
@@ -159,7 +158,7 @@ def evaluate_before(psm_list: PSMList, config: Dict) -> RescoreResult:
     )
 
 
-def evaluate_before_from_provenance(psm_list: PSMList, config: Dict) -> RescoreResult:
+def evaluate_before_from_provenance(psm_list: PSMList, config: dict) -> RescoreResult:
     """
     Rebuild the "before" ``RescoreResult`` for standalone report regeneration.
 
@@ -181,7 +180,7 @@ def evaluate_before_from_provenance(psm_list: PSMList, config: Dict) -> RescoreR
     return evaluate_before(psm_list, config)
 
 
-def evaluate_after_from_psm_list(psm_list: PSMList, config: Dict) -> RescoreResult:
+def evaluate_after_from_psm_list(psm_list: PSMList, config: dict) -> RescoreResult:
     """
     Rebuild the "after" ``RescoreResult`` for standalone report regeneration.
 

ms2rescore/_utils.py

@@ -4,23 +4,22 @@
 import os
 from glob import glob
 from pathlib import Path
-from typing import Optional, Union
 
 import numpy as np
 import pandas as pd
 from ms2rescore_rs import is_supported_file_type
 from psm_utils import PSMList
 
-from ms2rescore.exceptions import MS2RescoreConfigurationError
 from ms2rescore._ristretto_utils import _is_original_psm
+from ms2rescore.exceptions import MS2RescoreConfigurationError
 
 logger = logging.getLogger(__name__)
 
 
 def infer_spectrum_path(
-    configured_path: Union[str, Path, None],
-    run_name: Optional[str] = None,
-) -> Union[str, Path]:
+    configured_path: str | Path | None,
+    run_name: str | None = None,
+) -> str | Path:
     """
     Infer spectrum path from passed path and expected filename (e.g. from PSM file).
 

ms2rescore/_version.py

@@ -8,15 +8,13 @@
 import importlib.metadata
 import json
 import logging
+import tomllib as toml
 from pathlib import Path
-from typing import Dict, Optional, Tuple, Union
 from urllib.error import HTTPError, URLError
 from urllib.request import Request, urlopen
 
 from packaging.version import Version
 
-import tomllib as toml
-
 from ms2rescore.exceptions import MS2RescoreError
 
 LOGGER = logging.getLogger(__name__)
@@ -28,17 +26,16 @@
 class UpdateCheckError(MS2RescoreError):
     """An error occurred while checking for software updates."""
 
-    pass
 
 
-def _version_from_metadata() -> Optional[Version]:
+def _version_from_metadata() -> Version | None:
     try:
         return Version(importlib.metadata.version("ms2rescore"))
     except importlib.metadata.PackageNotFoundError:
         return None
 
 
-def _version_from_pyproject() -> Optional[Version]:
+def _version_from_pyproject() -> Version | None:
     pyproject = Path(__file__).resolve().parents[1] / "pyproject.toml"
     if not pyproject.is_file():
         return None
@@ -56,7 +53,7 @@ def _version_from_pyproject() -> Optional[Version]:
     return None
 
 
-def _get_latest_version(timeout_seconds: float) -> Tuple[Version, Optional[str]]:
+def _get_latest_version(timeout_seconds: float) -> tuple[Version, str | None]:
     """Check GitHub latest release and return the version string."""
     # Prepare GitHub API request
     url = f"https://api.github.com/repos/{_GITHUB_REPO}/releases/latest"
@@ -95,13 +92,13 @@ def get_version() -> str:
 
 
 def check_for_update(
-    timeout_seconds: Optional[float] = None,
-) -> Dict[str, Optional[Union[str, bool]]]:
+    timeout_seconds: float | None = None,
+) -> dict[str, str | bool | None]:
     """Check GitHub latest release and report whether an update exists."""
     timeout_seconds = timeout_seconds or _GITHUB_TIMEOUT_SECONDS
 
     # Initialize result dictionary
-    result: Dict[str, Optional[Union[str, bool]]] = {
+    result: dict[str, str | bool | None] = {
         "update_available": False,
         "current_version": None,
         "latest_version": None,
@@ -128,5 +125,6 @@ def check_for_update(
         result["update_available"] = latest_version > current_version
     except Exception:
         # If current_version can't be parsed, don't treat as updateable
+        LOGGER.exception("Update check failed")
         result["update_available"] = False
     return result

ms2rescore/config_parser.py

@@ -4,11 +4,9 @@
 import json
 import multiprocessing as mp
 import re
+import tomllib
 from argparse import Namespace
 from pathlib import Path
-from typing import Dict, List, Union
-
-import tomllib
 
 from cascade_config import CascadeConfig
 
@@ -36,7 +34,7 @@ def _parse_output_path(configured_path, psm_file_path):
         return (Path(psm_file_path).parent / psm_file_stem).as_posix()
 
 
-def _validate_filenames(config: Dict) -> Dict:
+def _validate_filenames(config: dict) -> dict:
     """Validate and infer input/output filenames."""
     # psm_file should be provided
     if not config["ms2rescore"]["psm_file"]:
@@ -74,7 +72,7 @@ def _validate_filenames(config: Dict) -> Dict:
     return config
 
 
-def _validate_processes(config: Dict) -> Dict:
+def _validate_processes(config: dict) -> dict:
     """Validate requested processes with available cpu count."""
     n_available = mp.cpu_count()
     if (config["ms2rescore"]["processes"] == -1) or (
@@ -84,7 +82,7 @@ def _validate_processes(config: Dict) -> Dict:
     return config
 
 
-def _validate_regular_expressions(config: Dict) -> Dict:
+def _validate_regular_expressions(config: dict) -> dict:
     """Validate regular expressions in configuration."""
     for field in [
         "psm_id_pattern",
@@ -113,7 +111,7 @@ def _validate_regular_expressions(config: Dict) -> Dict:
     return config
 
 
-def parse_configurations(configurations: List[Union[dict, str, Path, Namespace]]) -> Dict:
+def parse_configurations(configurations: list[dict | str | Path | Namespace]) -> dict:
     """
     Parse and validate MS²Rescore configuration files and CLI arguments.
 
@@ -150,19 +148,20 @@ def parse_configurations(configurations: List[Union[dict, str, Path, Namespace]]
             continue
         if isinstance(config, dict):
             cascade_conf.add_dict(config)
-        elif isinstance(config, str) or isinstance(config, Path):
+        elif isinstance(config, (str, Path)):
             if Path(config).suffix.lower() == ".json":
                 cascade_conf.add_json(config)
             elif Path(config).suffix.lower() == ".toml":
-                cascade_conf.add_dict(dict(tomllib.load(Path(config).open("rb"))))
+                with Path(config).open("rb") as f:
+                    cascade_conf.add_dict(dict(tomllib.load(f)))
             else:
                 raise MS2RescoreConfigurationError(
                     "Unknown file extension for configuration file. Should be `json` or `toml`."
                 )
         elif isinstance(config, Namespace):
             cascade_conf.add_namespace(config, subkey="ms2rescore")
         else:
-            raise ValueError(
+            raise TypeError(
                 "Configuration should be a dictionary, argparse Namespace, or path to a "
                 "configuration file."
             )

ms2rescore/core.py

@@ -1,7 +1,6 @@
 import json
 import logging
 from multiprocessing import cpu_count
-from typing import Dict, Optional
 
 import psm_utils.io
 from psm_utils import PSMList
@@ -16,7 +15,7 @@
 logger = logging.getLogger(__name__)
 
 
-def rescore(configuration: Dict, psm_list: Optional[PSMList] = None) -> None:
+def rescore(configuration: dict, psm_list: PSMList | None = None) -> None:
     """
     Run full MS²Rescore workflow with passed configuration.
 
@@ -55,11 +54,11 @@ def rescore(configuration: Dict, psm_list: Optional[PSMList] = None) -> None:
             )
 
     # Define feature names; get existing feature names from PSM file
-    feature_names = dict()
+    feature_names = {}
     psm_list_feature_names = {
         feature_name
         for psm_list_features in psm_list["rescoring_features"]
-        for feature_name in psm_list_features.keys()
+        for feature_name in psm_list_features
     }
     feature_names["psm_file"] = psm_list_feature_names
     logger.debug(
@@ -81,7 +80,7 @@ def rescore(configuration: Dict, psm_list: Optional[PSMList] = None) -> None:
     # Add missing precursor info from spectrum file if needed
     required_ms_data = {
         ms_data
-        for fgen_name in config["feature_generators"].keys()
+        for fgen_name in config["feature_generators"]
         if fgen_name not in skip_fgens
         for ms_data in FEATURE_GENERATORS[fgen_name].required_ms_data
     }
@@ -208,7 +207,7 @@ def rescore(configuration: Dict, psm_list: Optional[PSMList] = None) -> None:
 
     # Rename PSMs to USIs if requested, reusing the lookup built above
     if config["rename_to_usi"]:
-        logging.debug(f"Creating USIs for {len(psm_list)} PSMs")
+        logger.debug(f"Creating USIs for {len(psm_list)} PSMs")
         psm_list["spectrum_id"] = [usi_by_native_id[(psm.run, psm.spectrum_id)] for psm in psm_list]
 
     # Rescore PSMs
@@ -268,14 +267,13 @@ def rescore(configuration: Dict, psm_list: Optional[PSMList] = None) -> None:
                 fdr_threshold=config["report_fdr"],
             )
             generate.generate_report(output_file_root, report_data)
-        except exceptions.ReportGenerationError as e:
-            logger.exception(e)
+        except exceptions.ReportGenerationError:
+            logger.exception("Report generation failed")
 
 
 def _write_feature_names(feature_names, output_file_root):
     """Write feature names to file."""
     with open(output_file_root + ".feature_names.tsv", "w") as f:
         f.write("feature_generator\tfeature_name\n")
         for fgen, fgen_features in feature_names.items():
-            for feature in fgen_features:
-                f.write(f"{fgen}\t{feature}\n")
+            f.writelines(f"{fgen}\t{feature}\n" for feature in fgen_features)

ms2rescore/exceptions.py

@@ -4,46 +4,38 @@
 class MS2RescoreError(Exception):
     """Generic MS2Rescore error."""
 
-    pass
 
 
 class MS2RescoreConfigurationError(MS2RescoreError):
     """Invalid MS2Rescore configuration."""
 
-    pass
 
 
 class IDFileParsingError(MS2RescoreError):
     """Identification file parsing error."""
 
-    pass
 
 
 class ModificationParsingError(IDFileParsingError):
     """Identification file parsing error."""
 
-    pass
 
 
 class MissingValuesError(MS2RescoreError):
     """Missing values in PSMs and/or spectra."""
 
-    pass
 
 
 class ReportGenerationError(MS2RescoreError):
     """Error while generating report."""
 
-    pass
 
 
 class RescoringError(MS2RescoreError):
     """Error while rescoring PSMs."""
 
-    pass
 
 
 class ParseSpectrumError(MS2RescoreError):
     """Error while parsing spectrum files."""
 
-    pass