Update ruff rules

.pre-commit-config.yaml

@@ -56,6 +56,11 @@ repos:
           - pytest
           - types-PyYAML
 
+  - repo: https://github.com/crate-ci/typos
+    rev: v1.29.4
+    hooks:
+      - id: typos
+
   - repo: https://github.com/codespell-project/codespell
     rev: "v2.3.0"
     hooks:

CHANGELOG.md

@@ -22,7 +22,7 @@
 - Added a `timelimit_on_context` helper to raise an error after some specified
   length of time. Looking at you, BANE and issue #186. Arrrr.
 - Added a `BANE` callback handler to attempt to help #186. This includes a
-  `AttemptRerunException` and corresponding code in `run_singularity_comand` to
+  `AttemptRerunException` and corresponding code in `run_singularity_command` to
   retry the failing command.
 
 # 0.2.8

README.md

@@ -5,7 +5,7 @@
 
 A pirate themed toy ASKAP-RACS pipeline.
 
-Yarrrr-Harrrr fiddley-dee!
+Yarrrr-Harrrr fiddly-dee!
 
 <img src="docs/logo.jpeg" alt="Capn' Flint - Credit: DALLE 3" style="width:400px;"/>
 
@@ -148,7 +148,7 @@ the time of writing there are six containers for:
 - calibration: this should contain `calibrate` and `applysolutions`. These are
   tools written by Andre Offringa.
 - flagging: this should contain `aoflagger`, which is installable via a
-  `apt install aoflagger` within ubunutu.
+  `apt install aoflagger` within ubuntu.
 - imaging: this should contain `wsclean`. This should be at least version 3. At
   the moment a modified version is being used (which implements a
   `-force-mask-round` option).

docs/conf.py

@@ -5,6 +5,7 @@
 
 # -- Project information -----------------------------------------------------
 # https://www.sphinx-doc.org/en/master/usage/configuration.html#project-information
+from __future__ import annotations
 
 project = "flint"
 copyright = "2023, Tim Galvin"

docs/index.rst

@@ -8,7 +8,7 @@ Welcome to flint's documentation!
 
 A pirate themed toy ASKAP-RACS pipeline.
 
-Yarrrr-Harrrr fiddley-dee!
+Yarrrr-Harrrr fiddly-dee!
 
 .. image:: logo.jpeg
    :width: 400

flint/archive.py

@@ -1,13 +1,15 @@
 """Operations around preserving files and products from an flint run"""
 
+from __future__ import annotations
+
 import re
+import shlex
 import shutil
 import subprocess
-import shlex
 import tarfile
 from argparse import ArgumentParser
 from pathlib import Path
-from typing import Any, Collection, Dict, List, Tuple
+from typing import Any, Collection
 
 from flint.configuration import get_options_from_strategy
 from flint.exceptions import TarArchiveError
@@ -21,7 +23,7 @@
 
 def resolve_glob_expressions(
     base_path: Path, file_re_patterns: Collection[str]
-) -> Tuple[Path, ...]:
+) -> tuple[Path, ...]:
     """Collect a set of files given a base directory and a set of glob expressions. Unique
     paths are returned.
 
@@ -34,7 +36,7 @@
     """
     base_path = Path(base_path)
 
-    resolved_files: List[Path] = []
+    resolved_files: list[Path] = []
 
     logger.info(f"Searching {base_path=}")
 
@@ -70,7 +72,7 @@
 
     copy_out_path.mkdir(parents=True, exist_ok=True)
     total = len(files_to_copy)
-    not_copied: List[Path] = []
+    not_copied: list[Path] = []
 
     logger.info(f"Copying {total} files into {copy_out_path}")
     for count, file in enumerate(files_to_copy):
@@ -111,7 +113,7 @@
     ), f"{tarball} is not a file or does not exist"
     assert tarball.suffix == ".tar", f"{tarball=} appears to not have a .tar extension"
 
-    cmd = f"tar -tvf {str(tarball)}"
+    cmd = f"tar -tvf {tarball!s}"
     logger.info(f"Verifying {tarball=}")
     popen = subprocess.Popen(shlex.split(cmd), stderr=subprocess.PIPE)
     with popen.stderr:  # type: ignore
@@ -153,7 +155,7 @@
     logger.info(f"Opening {tar_out_path}")
     with tarfile.open(tar_out_path, "w") as tar:
         for count, file in enumerate(files_to_tar):
-            logger.info(f"{count+1} of {total}, adding {str(file)}")
+            logger.info(f"{count+1} of {total}, adding {file!s}")
             tar.add(file, arcname=file.name)
 
     logger.info(f"Created {tar_out_path}")
@@ -217,7 +219,7 @@
     return copy_out_path
 
 
-def get_archive_options_from_yaml(strategy_yaml_path: Path) -> Dict[str, Any]:
+def get_archive_options_from_yaml(strategy_yaml_path: Path) -> dict[str, Any]:
     """Load the archive options from a specified strategy file
 
     Args:
@@ -360,7 +362,7 @@
             )
         )
 
-        update_options_create: Dict[str, Any] = (
+        update_options_create: dict[str, Any] = (
             get_archive_options_from_yaml(strategy_yaml_path=args.strategy_yaml_path)
             if args.strategy_yaml_path
             else dict(tar_file_re_patterhs=args.file_patterns)

flint/bandpass.py

@@ -1,21 +1,22 @@
 """Procedure to calibrate bandpass observation"""
 
+from __future__ import annotations
+
 from argparse import ArgumentParser
 from pathlib import Path
-from typing import Optional, Union
 
 import numpy as np
 from casacore.tables import table, taql
 
 from flint.calibrate.aocalibrate import AOSolutions, calibrate_apply_ms
 from flint.flagging import flag_ms_aoflagger
 from flint.logging import logger
-from flint.ms import MS, describe_ms, preprocess_askap_ms, get_field_id_for_field
+from flint.ms import MS, describe_ms, get_field_id_for_field, preprocess_askap_ms
 from flint.naming import create_ms_name
 from flint.sky_model import KNOWN_1934_FILES, get_1934_model
 
 
-def plot_solutions(solutions_path: Path, ref_ant: Optional[int] = 0) -> None:
+def plot_solutions(solutions_path: Path, ref_ant: int | None = 0) -> None:
     """Plot solutions for AO-style solutions
 
     Args:
@@ -28,7 +29,7 @@
     _ = ao_sols.plot_solutions(ref_ant=ref_ant)
 
 
-def flag_bandpass_offset_pointings(ms: Union[MS, Path]) -> MS:
+def flag_bandpass_offset_pointings(ms: MS | Path) -> MS:
     """The typical bandpass style observation in ASKAP will shift each beam
     so that it is centred on the bandpass-calibration object (here B1934-638).
     During each offset position all beams are recording data still. The trick
@@ -61,7 +62,7 @@
     logger.info(f"The B1934-638 field name is {good_field_name}. ")
     logger.info("Will attempt to flag other fields. ")
 
-    with table(f"{str(ms.path)}/FIELD", readonly=True, ack=False) as tab:
+    with table(f"{ms.path!s}/FIELD", readonly=True, ack=False) as tab:
         # The ID is _position_ of the matching row in the table.
         field_names = tab.getcol("NAME")
         field_idx = np.argwhere([fn == good_field_name for fn in field_names])[0]
@@ -73,7 +74,7 @@
         field_idx = field_idx[0]
         logger.info(f"{good_field_name} FIELD_ID is {field_idx}")
 
-    with table(f"{str(ms.path)}", readonly=False, ack=False) as tab:
+    with table(f"{ms.path!s}", readonly=False, ack=False) as tab:
         field_idxs = tab.getcol("FIELD_ID")
         field_mask = field_idxs != field_idx
         logger.info(
@@ -92,9 +93,9 @@
 
 
 def extract_correct_bandpass_pointing(
-    ms: Union[MS, Path],
+    ms: MS | Path,
     source_name_prefix: str = "B1934-638",
-    ms_out_dir: Optional[Path] = None,
+    ms_out_dir: Path | None = None,
 ) -> MS:
     """The typical bandpass style observation in ASKAP will shift each beam
     so that it is centred on the bandpass-calibration object (here B1934-638).
@@ -134,7 +135,7 @@
     ms = MS.cast(ms)
     ms_summary = describe_ms(ms, verbose=False)
 
-    logger.info(f"Checking for unique fields in {str(ms.path)} data table.")
+    logger.info(f"Checking for unique fields in {ms.path!s} data table.")
     with table(str(ms.path)) as tab:
         fields = np.unique(tab.getcol("FIELD_ID"))
         if len(fields) == 1:
@@ -164,7 +165,7 @@
         out_path = ms_out_dir / Path(out_name).name
 
     logger.info(f"Will create a MS, writing to {out_path}")
-    with table(f"{str(ms.path)}") as tab:
+    with table(f"{ms.path!s}") as tab:
         field_ms = taql(f"select * from $tab where FIELD_ID=={field_id}")
         field_ms.copy(str(out_path), deep=True)
 
@@ -177,8 +178,8 @@
     mode: str,
     calibrate_container: Path,
     plot: bool = True,
-    aoflagger_container: Optional[Path] = None,
-    ms_out_dir: Optional[Path] = None,
+    aoflagger_container: Path | None = None,
+    ms_out_dir: Path | None = None,
 ) -> MS:
     """Entry point to extract the appropriate field from a bandpass observation,
     run AO-style calibrate, and plot results. In its current form a new measurement
@@ -196,7 +197,7 @@
     Returns:
         MS: The calibrated measurement set with nominated column
     """
-    logger.info(f"Will calibrate {str(ms_path)}, column {data_column}")
+    logger.info(f"Will calibrate {ms_path!s}, column {data_column}")
 
     # TODO: Check to make sure only 1934-638
     model_path: Path = get_1934_model(mode=mode)

flint/bptools/preflagger.py

@@ -6,8 +6,10 @@
 components of the bandpass.
 """
 
+from __future__ import annotations
+
 from pathlib import Path
-from typing import List, NamedTuple, Optional, Tuple
+from typing import NamedTuple
 
 import matplotlib.pyplot as plt
 import numpy as np
@@ -32,9 +34,9 @@
     """The initial model of the complex_gains"""
     fit_model_gains: np.ndarray
     """The complex gain model fit made against the unwrapped gains (i.e. complex_gains / init_model_gains)"""
-    init_model_params: Tuple[float, float]
+    init_model_params: tuple[float, float]
     """The initial guess (gradient, offset) model parameters to represent the phase component of the complex_gains"""
-    fit_model_params: Tuple[float, float]
+    fit_model_params: tuple[float, float]
     """The fitted model parameters constrained against the unwrapped gains"""
     outlier_mask: np.ndarray
     """Boolean mask of equal length to complex_gain, where True represents outliers that should be flagged"""
@@ -50,7 +52,7 @@
 def plot_phase_outlier(
     phase_outlier_results: PhaseOutlierResults,
     output_path: Path,
-    title: Optional[str] = None,
+    title: str | None = None,
 ) -> Path:
     """Create a simple diagnostic plot highlighting how the outlier
     channels and their phases were selected.
@@ -63,7 +65,7 @@
     Returns:
         Path: Path of the output image file
     """
-    logger.debug(f"Creating phase outlier plot, writing {str(output_path)}.")
+    logger.debug(f"Creating phase outlier plot, writing {output_path!s}.")
 
     complex_gains = phase_outlier_results.complex_gains
     init_model_gains = phase_outlier_results.init_model_gains
@@ -165,8 +167,8 @@
     complex_gains: np.ndarray,
     flag_cut: float,
     use_mad: bool = False,
-    plot_title: Optional[str] = None,
-    plot_path: Optional[Path] = None,
+    plot_title: str | None = None,
+    plot_path: Path | None = None,
 ) -> PhaseOutlierResults:
     """This procedure attempts to identify channels in the bandpass solutions to
     flag but searching for gains with outlier phases. Typically, ASKAP solutions
@@ -295,7 +297,7 @@
 
 
 def flags_over_threshold(
-    flags: np.ndarray, thresh: float = 0.8, ant_idx: Optional[int] = None
+    flags: np.ndarray, thresh: float = 0.8, ant_idx: int | None = None
 ) -> bool:
     """Given a set of flags for an antenna across frequency, consider how much is flagged, indicated
     by a value of True, and return whether it was over a threshold. The intent is to return whether
@@ -333,7 +335,7 @@
     mean: float,
     std: float,
     output_path: Path,
-    plot_title: Optional[str] = None,
+    plot_title: str | None = None,
 ) -> Path:
     """A simply plot to examine the polynomial fit to the residual amplitude data.
 
@@ -378,8 +380,8 @@
     complex_gains: np.ndarray,
     use_robust: bool = True,
     polynomial_order: int = 5,
-    plot_path: Optional[Path] = None,
-    plot_title: Optional[str] = None,
+    plot_path: Path | None = None,
+    plot_title: str | None = None,
 ) -> bool:
     """Calculate the median or mean of the residual amplitudes of the complex gains
     after fitting a polynomial of order polynomial_order.
@@ -520,7 +522,7 @@
     nant = mask.shape[0]
     logger.info(f"Accumulating flagged channels over {nant=} antenna")
 
-    empty_ants: List[int] = []
+    empty_ants: list[int] = []
 
     # TODO: This can be replaced with numpy broadcasting
 

flint/bptools/smoother.py

@@ -1,4 +1,4 @@
-from typing import Tuple
+from __future__ import annotations
 
 import numpy as np
 from scipy.ndimage import median_filter
@@ -207,7 +207,7 @@ def smooth_bandpass_complex_gains(
     window_size: int = 16,
     polynomial_order: int = 4,
     apply_median_filter: bool = True,
-    smooth_jones_elements: Tuple[int, ...] = (0, 1, 2, 3),
+    smooth_jones_elements: tuple[int, ...] = (0, 1, 2, 3),
 ) -> np.ndarray:
     """Smooth bandpass solutions by applying a savgol filter to the real and imaginary components
     of each of the antenna based polarisation solutions across channels.