Enable ty rules: not-iterable, call-non-callable, unsupported-operator, invalid-assignment

dascore/compat.py

@@ -9,6 +9,7 @@
 
 import importlib
 from contextlib import suppress
+from typing import TypeGuard
 
 import numpy as np
 from h5py import Dataset as H5Dataset
@@ -94,7 +95,7 @@ def array(array):
     return _make_immutable(out)
 
 
-def is_array(maybe_array):
+def is_array(maybe_array) -> TypeGuard[np.ndarray]:
     """
     Determine if an object is a numpy array.
     """

dascore/core/attrs.py

@@ -136,8 +136,8 @@ def from_dict(
             return attr_map
         if attr_map is None:
             out = {}
-        elif hasattr(attr_map, "model_dump"):
-            out = attr_map.model_dump()
+        elif callable(model_dump := getattr(attr_map, "model_dump", None)):
+            out = model_dump()
         else:
             out = attr_map
         if isinstance(out, Mapping):

dascore/core/coordmanager.py

@@ -45,7 +45,7 @@
 from collections.abc import Mapping, Sequence
 from itertools import zip_longest
 from types import EllipsisType
-from typing import Annotated
+from typing import Annotated, Any
 
 import numpy as np
 from pydantic import field_validator, model_validator
@@ -300,9 +300,8 @@ def _divide_kwargs(kwargs):
         # update based on keywords
         for item, value in coord_updates.items():
             coord_name, attr = item.split("_")
-            new = list(out[coord_name])
-            new[1] = new[1].update(**{attr: value})
-            out[coord_name] = tuple(new)
+            coord_dims, coord = out[coord_name]
+            out[coord_name] = (coord_dims, coord.update(**{attr: value}))
 
         dims = tuple(x for x in dims if x not in coord_to_drop)
         return get_coord_manager(out, dims=dims)
@@ -1000,15 +999,15 @@ def _get_coord_dims_tuple(self):
         dim_map = self.dim_map
         return tuple((name, *dim_map[name]) for name in self.coord_map)
 
-    def _get_indexer(self, ind: int | None = None, value=None):
+    def _get_indexer(self, ind: int, value=None):
         """
         Get an indexer for the appropriate data shape.
 
         This is useful for generating a tuple that can be used
 
         ind is a list of indices to substitute in values.
         """
-        out = [slice(None, None) for _ in self.shape]
+        out: list[Any] = [slice(None, None) for _ in self.shape]
         out[ind] = value
         return tuple(out)
 

dascore/core/coords.py

@@ -18,7 +18,7 @@
 from functools import cache
 from operator import gt, lt
 from types import EllipsisType
-from typing import Any, Literal
+from typing import Any, Literal, overload
 
 import numpy as np
 import pandas as pd
@@ -481,9 +481,15 @@ def valid_non_coord(coord1, coord2):
         coord2, slice2 = other.order(intersection)
         return coord1, coord2, slice1, slice2
 
+    @overload
+    def __getitem__(self, item: int | np.integer) -> Any: ...
+
+    @overload
+    def __getitem__(self, item: slice | np.ndarray) -> Self: ...
+
     @abc.abstractmethod
-    def __getitem__(self, item) -> Self:
-        """Should implement slicing and return new instance."""
+    def __getitem__(self, item):
+        """Index the coord; slices return a new coord, int indices a value."""
 
     @cached_method
     def __len__(self):
@@ -1139,17 +1145,17 @@ def reduce_coord(self, dim_reduce="empty"):
             new_coord = get_coord(shape=(1,), units=self.units, dtype=self.dtype)
         elif dim_reduce == "squeeze":
             return None
-        elif (func := _AGG_FUNCS.get(dim_reduce)) or callable(dim_reduce):
-            func = dim_reduce if callable(dim_reduce) else func
+        else:
+            func = dim_reduce if callable(dim_reduce) else _AGG_FUNCS.get(dim_reduce)
+            if func is None:
+                msg = "dim_reduce must be 'empty', 'squeeze' or valid aggregator."
+                raise ParameterError(msg)
             coord_data = self.data
             if dtype_time_like(coord_data):
                 result = _reduce_time_like(func, coord_data)
             else:
                 result = func(self.data)
             new_coord = self.update(data=result)
-        else:
-            msg = "dim_reduce must be 'empty', 'squeeze' or valid aggregator."
-            raise ParameterError(msg)
         return new_coord
 
 
@@ -2258,6 +2264,7 @@ def select(
             kept.append(sub)
         if not kept:
             return self.empty(), slice(0, 0)
+        assert hi is not None  # kept is non-empty, so the loop set hi
         new = self._rebuild(kept)
         start = None if lo == 0 else lo
         stop = None if hi >= len(self) else hi

dascore/core/spool.py

@@ -539,6 +539,7 @@ def __getitem__(self, item) -> PatchType | BaseSpool:
     def __iter__(self):
         # The catalog snapshots the relation once and skips patches which
         # cannot be resolved (see #583).
+        assert self._catalog is not None  # __init__ always sets the catalog
         yield from self._catalog
 
     # --- selection and presentation specs -------------------------------

dascore/io/index/backend.py

@@ -886,7 +886,7 @@ def _flatten(
                 if kinds == {"str"}:
                     # flat-contract convention: missing strings are ""
                     series = series.fillna("")
-                new_columns[name] = series
+                new_columns[str(name)] = series
         if cols_to_drop:
             out = out.drop(columns=cols_to_drop)
         # flat-contract names for source columns; path_attrs goes private

dascore/io/index/indexer.py

@@ -127,7 +127,7 @@ class DBDirectoryIndexer:
 
     def __init__(
         self,
-        path: str | Path,
+        path: str | Path | UPath,
         index_path: str | Path | None = None,
     ):
         path = UPath(path).absolute() if isinstance(path, UPath) else Path(path)

dascore/io/index/planned.py

@@ -125,7 +125,9 @@ def _coord_record_from_row(
         units = None
     length = None
     if step is not None:
-        length = int(round((hi - lo) / step)) + 1
+        # lo, hi, and step always share a time kind (or are all floats), but
+        # ty unions the branch types and rejects the mixed combinations.
+        length = int(round((hi - lo) / step)) + 1  # ty: ignore[unsupported-operator]
     key = row.get(f"_{name}_def_key")
     fingerprint = None
     if isinstance(key, str) and key.startswith("fp:"):

dascore/io/segy/core.py

@@ -41,10 +41,10 @@ def read(self, path: LocalPath, time=None, channel=None, **kwargs):
         be implemented as well.
         """
         segyio = optional_import(self._package_name)
-        path = str(path)
-        with segyio.open(path, ignore_geometry=True) as fi:
+        path_str = str(path)
+        with segyio.open(path_str, ignore_geometry=True) as fi:
             coords = _get_coords(fi)
-            attrs = _get_attrs(fi, coords, path, self, include_source=True)
+            attrs = _get_attrs(fi, coords, path_str, self, include_source=True)
             data, coords = _get_filtered_data_and_coords(
                 fi, coords, time=time, channel=channel
             )
@@ -61,10 +61,10 @@ def scan(self, path: LocalPath, **kwargs) -> list[ScanPayload]:
         Returns lightweight scan metadata without loading the data array.
         """
         segyio = optional_import(self._package_name)
-        path = str(path)
-        with segyio.open(path, ignore_geometry=True) as fi:
+        path_str = str(path)
+        with segyio.open(path_str, ignore_geometry=True) as fi:
             coords = _get_coords(fi)
-            attrs = _get_attrs(fi, coords, path, self)
+            attrs = _get_attrs(fi, coords, path_str, self)
             dtype = str(fi.dtype)
         return [
             {

dascore/io/sintela/utils.py

@@ -103,7 +103,9 @@ def _read_base_header(fid):
     """Return the first 3 elements of the sintela header."""
     data = fid.read(base_header_dtypes.itemsize)
     array = np.frombuffer(data, dtype=base_header_dtypes, count=1)
-    out = {x: y for x, y in zip(array.dtype.names, array[0])}
+    names = array.dtype.names
+    assert names is not None  # structured dtypes always have field names
+    out = {x: y for x, y in zip(names, array[0])}
     return out
 
 
@@ -130,7 +132,9 @@ def _read_remaining_header(fid, base):
     dtype = _HEADER_DTYPES[version]
     data = fid.read(dtype.itemsize)
     buf = np.frombuffer(data, dtype=dtype, count=1)
-    header = {x: y for x, y in zip(buf.dtype.names, buf[0])}
+    names = buf.dtype.names
+    assert names is not None  # structured dtypes always have field names
+    header = {x: y for x, y in zip(names, buf[0])}
     assert version == "3", "only 3 support for now,"
     header["num_packets"] = _get_number_of_packets(fid, header, header_size)
     header["dtype"] = "<f4"

dascore/io/tdms/utils.py

@@ -5,6 +5,7 @@
 import datetime
 import mmap
 import struct
+from typing import Any
 
 import numpy as np
 
@@ -176,7 +177,7 @@ def _get_all_attrs(tdms_file, lead_in_length=28):
     # lead_in is 28 bytes:
     fields = struct.unpack("<4siiQQ", lead_in)
     # Keep track of information about file in fileinfo
-    fileinfo = dict(zip(FILEINFO_NAMES, fields))
+    fileinfo: dict[str, Any] = dict(zip(FILEINFO_NAMES, fields))
     fileinfo["decimated"] = not bool(fileinfo["toc"] & DECIMATE_MASK)
     # Make offsets relative to beginning of file:
     fileinfo["next_segment_offset"] += lead_in_length
@@ -218,7 +219,7 @@ def _get_all_attrs(tdms_file, lead_in_length=28):
     tdms_file.seek(var + 4, 1)
     fileinfo["data_type"] = TDS_DATA_TYPE.get(struct.unpack("<i", tdms_file.read(4))[0])
     if fileinfo["data_type"] not in ("int16", "float32"):
-        raise Exception("Unsupported TDMS data type: " + fileinfo["data_type"])
+        raise Exception(f"Unsupported TDMS data type: {fileinfo['data_type']}")
     # get number of samples by dividing amount of unread data by the
     # size of data per channel
     numofsamples = (

dascore/io/wav/core.py

@@ -7,6 +7,7 @@
 import numpy as np
 from scipy.io.wavfile import write
 
+from dascore.compat import UPath
 from dascore.constants import ONE_SECOND, SpoolType
 from dascore.exceptions import ParameterError
 from dascore.io.core import FiberIO
@@ -20,7 +21,11 @@ class WavIO(FiberIO):
     name = "WAV"
 
     def write(
-        self, spool: SpoolType, resource: str | Path, resample_frequency=None, **kwargs
+        self,
+        spool: SpoolType,
+        resource: str | Path | UPath,
+        resample_frequency=None,
+        **kwargs,
     ):
         """
         Write the contents of the patch to one or more wav files.

dascore/proc/coords.py

@@ -159,7 +159,7 @@ def get_array(
     name: str | None = None,
     require_sorted: bool = False,
     require_evenly_sampled: bool = False,
-) -> BaseCoord:
+) -> np.ndarray:
     """
     Get an array associated with patch data or a coordinate.
 

dascore/proc/filter.py

@@ -596,7 +596,7 @@ def _maybe_transform_units(filt, dft_patch, freq_dims):
             filt = filt * dc.get_quantity(units)
         if not isinstance(filt, dc.units.Quantity):
             return filt
-        array, units = filt.magnitude, filt.units
+        array, units = np.asarray(filt.magnitude), filt.units
         coord_unit_1 = dft_patch.get_coord(freq_dims[-1]).units
         coord_unit_2 = dft_patch.get_coord(freq_dims[-2]).units
         if not (coord_unit_1 and coord_unit_2):

dascore/proc/mute.py

@@ -4,7 +4,7 @@
 
 from abc import ABC, abstractmethod
 from collections.abc import Mapping, Sized
-from typing import ClassVar
+from typing import Any, ClassVar
 
 import numpy as np
 from numpy.linalg import norm
@@ -137,7 +137,7 @@ def from_params(cls, vals, dims, axes, patch, relative):
     def _apply_mask(self, array: NDArray, patch: dc.Patch, fill_value) -> NDArray:
         coord = patch.get_coord(self.dims[0])
         _, c_index = coord.select(self.lims, relative=self.relative)
-        index = [slice(None)] * array.ndim
+        index: list[Any] = [slice(None)] * array.ndim
         index[self.axes[0]] = c_index
         array[tuple(index)] = fill_value
         return array

dascore/transform/fbe.py

@@ -8,7 +8,6 @@
 from dascore.units import get_filter_units
 from dascore.utils.misc import check_filter_kwargs, check_filter_range
 from dascore.utils.patch import get_dim_sampling_rate, patch_function
-from dascore.utils.time import to_float
 
 
 @patch_function()
@@ -84,7 +83,7 @@ def fbe(
     check_filter_range(nyquist, low, high, filt_min, filt_max)
 
     if step is None:
-        step = to_float(1 / sample_rate)
+        step = 1 / sample_rate
 
     patch = patch.pass_filter(**kwargs)
 

dascore/transform/fourier.py

@@ -282,11 +282,11 @@ def dft(
     >>> # calculate a power spectral density along time
     >>> psd = patch.dft(dim="time", real=True, output="PSD")
     """
-    output = output.upper()
-    if output not in DFT_OUTPUT_TYPES:
+    output_type = output.upper()
+    if output_type not in DFT_OUTPUT_TYPES:
         msg = f"Unknown output={output!r}. Expected one of: {DFT_OUTPUT_TYPES}."
         raise ValueError(msg)
-    if output == "FFT" and db:
+    if output_type == "FFT" and db:
         msg = "db=True is only supported for output='AS', 'PS', or 'PSD'."
         raise ParameterError(msg)
 
@@ -318,11 +318,13 @@ def dft(
     shift_slice = slice(None) if real is None else slice(None, -1)
     data = nft.fftshift(fft_data, axes=axes[shift_slice])
     # get attributes
-    attrs = _get_dft_attrs(patch, dims, new_coords, pad=pad, output=output)
+    attrs = _get_dft_attrs(patch, dims, new_coords, pad=pad, output=output_type)
     patch_out = patch.new(data=data, coords=new_coords, attrs=attrs)
 
-    if output != "FFT":
-        patch_out = _convert_dft_spectral_amplitudes(patch_out, output, dims, real, db)
+    if output_type != "FFT":
+        patch_out = _convert_dft_spectral_amplitudes(
+            patch_out, output_type, dims, real, db
+        )
 
     return patch_out