Merge pull request #284 from scipp/type-hints

Fix some type hints and run mypy

Author: jl-wynen

.github/workflows/test.yml

@@ -77,3 +77,18 @@ jobs:
         with:
           name: CoverageReport
           path: coverage_html/
+
+  mypy:
+    runs-on: ${{ inputs.os-variant }}
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          ref: ${{ inputs.checkout_ref }}
+      - uses: actions/setup-python@v5
+        with:
+          python-version: ${{ inputs.python-version }}
+      - run: python -m pip install --upgrade pip
+      # Use mypy manually because it does not respect the exclusion patterns
+      # when called through tox.
+      - run: python -m pip install -r requirements/mypy.txt -e .
+      - run: python -m mypy .

docs/conf.py

@@ -217,7 +217,7 @@
 # relative to this directory. They are copied after the builtin static files,
 # so a file named "default.css" will overwrite the builtin "default.css".
 html_static_path = ['_static']
-html_css_files = []
+html_css_files: list[str] = []
 html_js_files = ["anaconda-icon.js"]
 
 # -- Options for HTMLHelp output ------------------------------------------

docs/user-guide/nexus-classes.ipynb

@@ -270,7 +270,8 @@
    "metadata": {},
    "source": [
     "In some cases the event data fields may be contained directly within an [NXdetector](#NXdetector).\n",
-    "The event data can also be accessed from there:"
+    "The event data can also be accessed from there:\n",
+    "(This example only shows one of several datasets that comprise event data.)"
    ]
   },
   {
@@ -280,7 +281,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "f['entry/instrument/bank102']['events'][...]"
+    "f['entry/instrument/bank102']['event_time_offset'][:1000]"
    ]
   },
   {

pyproject.toml

@@ -117,6 +117,24 @@ enable_error_code = [
 ]
 warn_unreachable = true
 
+exclude = [
+    # not yet updated for type hints:
+    '^tests/',
+    '^src/scippnexus/nxlog',
+    '^src/scippnexus/nxsample',
+    '^src/scippnexus/nxoff_geometry',
+    '^src/scippnexus/nxevent_data',
+    '^src/scippnexus/event_field',
+    '^src/scippnexus/nxcylindrical_geometry',
+    '^src/scippnexus/nxdata',
+    '^src/scippnexus/_load',
+    '^src/scippnexus/nxmonitor',
+    '^src/scippnexus/nxdetector',
+    '^src/scippnexus/application_definitions/',
+    '^src/scippnexus/nxtransformations',
+    '^src/scippnexus/base',
+]
+
 [tool.codespell]
 ignore-words-list = [
     # Codespell wants "socioeconomic" which seems to be the standard spelling.

src/scippnexus/__init__.py

@@ -11,6 +11,7 @@
 
 from . import typing
 from ._load import load
+from .attrs import Attrs
 from .base import (
     Group,
     NexusStructureError,
@@ -19,7 +20,7 @@
     create_class,
     create_field,
 )
-from .field import Attrs, DependsOn, Field
+from .field import DependsOn, Field
 from .file import File
 from .nexus_classes import *  # noqa: F403
 from .nxtransformations import TransformationChain, compute_positions, zip_pixel_offsets

src/scippnexus/_cache.py

@@ -9,18 +9,23 @@
 """
 
 # flake8: noqa: E501
+from __future__ import annotations
+
+from collections.abc import Callable
 from types import GenericAlias
+from typing import Generic, TypeVar
 
 _NOT_FOUND = object()
+R = TypeVar("R")
 
 
-class cached_property:
-    def __init__(self, func):
+class cached_property(Generic[R]):
+    def __init__(self, func: Callable[..., R]) -> None:
         self.func = func
-        self.attrname = None
+        self.attrname: str | None = None
         self.__doc__ = func.__doc__
 
-    def __set_name__(self, owner, name):
+    def __set_name__(self, owner: object, name: str) -> None:
         if self.attrname is None:
             self.attrname = name
         elif name != self.attrname:
@@ -29,9 +34,9 @@ def __set_name__(self, owner, name):
                 f"({self.attrname!r} and {name!r})."
             )
 
-    def __get__(self, instance, owner=None):
+    def __get__(self, instance: object, owner: object = None) -> R:
         if instance is None:
-            return self
+            return self  # type: ignore[return-value]
         if self.attrname is None:
             raise TypeError(
                 "Cannot use cached_property instance without calling __set_name__ on it."
@@ -46,7 +51,7 @@ def __get__(self, instance, owner=None):
                 f"instance to cache {self.attrname!r} property."
             )
             raise TypeError(msg) from None
-        val = cache.get(self.attrname, _NOT_FOUND)
+        val: R = cache.get(self.attrname, _NOT_FOUND)
         if val is _NOT_FOUND:
             val = self.func(instance)
             try:
@@ -59,4 +64,4 @@ def __get__(self, instance, owner=None):
                 raise TypeError(msg) from None
         return val
 
-    __class_getitem__ = classmethod(GenericAlias)
+    __class_getitem__ = classmethod(GenericAlias)  # type: ignore[var-annotated]

src/scippnexus/_common.py

@@ -2,6 +2,8 @@
 # Copyright (c) 2023 Scipp contributors (https://github.com/scipp)
 # @author Simon Heybrock
 
+from collections.abc import Sequence
+
 import numpy as np
 import scipp as sc
 
@@ -10,8 +12,8 @@
 
 def convert_time_to_datetime64(
     raw_times: sc.Variable,
-    start: str | None = None,
-    scaling_factor: float | np.float64 = None,
+    start: sc.Variable,
+    scaling_factor: float | np.float64 | None = None,
 ) -> sc.Variable:
     """
     The nexus standard allows an arbitrary scaling factor to be inserted
@@ -24,14 +26,16 @@ def convert_time_to_datetime64(
 
     See https://manual.nexusformat.org/classes/base_classes/NXlog.html
 
-    Args:
-        raw_times: The raw time data from a nexus file.
-        start: Optional, the start time of the log in an ISO8601
-            string. If not provided, defaults to the beginning of the
-            unix epoch (1970-01-01T00:00:00).
-        scaling_factor: Optional, the scaling factor between the provided
-            time series data and the unit of the raw_times Variable. If
-            not provided, defaults to 1 (a no-op scaling factor).
+    Parameters
+    ----------
+    raw_times:
+        The raw time data from a nexus file.
+    start:
+        The start time of the log.
+    scaling_factor:
+        Optional, the scaling factor between the provided
+        time series data and the unit of the raw_times Variable. If
+        not provided, defaults to 1 (a no-op scaling factor).
     """
     if (
         raw_times.dtype in (sc.DType.float64, sc.DType.float32)
@@ -53,10 +57,18 @@ def convert_time_to_datetime64(
     )
 
 
-def _to_canonical_select(dims: list[str], select: ScippIndex) -> dict[str, int | slice]:
+def has_time_unit(obj: sc.Variable) -> bool:
+    if (unit := obj.unit) is None:
+        return False
+    return unit.to_dict().get('powers') == {'s': 1}
+
+
+def to_canonical_select(
+    dims: Sequence[str], select: ScippIndex
+) -> dict[str, int | slice]:
     """Return selection as dict with explicit dim labels"""
 
-    def check_1d():
+    def check_1d() -> None:
         if len(dims) != 1:
             raise sc.DimensionError(
                 f"Dataset has multiple dimensions {dims}, "
@@ -67,8 +79,8 @@ def check_1d():
         return {}
     if isinstance(select, tuple) and len(select) == 0:
         return {}
-    if isinstance(select, tuple) and isinstance(select[0], str):
-        key, sel = select
+    if isinstance(select, tuple) and isinstance(select[0], str):  # type: ignore[misc]  # incorrect narrowing
+        key, sel = select  # type: ignore[misc]  # incorrect narrowing
         return {key: sel}
     if isinstance(select, tuple):
         check_1d()
@@ -77,7 +89,7 @@ def check_1d():
                 f"Dataset has single dimension {dims}, "
                 "but multiple indices {select} were specified."
             )
-        return {dims[0]: select[0]}
+        return {dims[0]: select[0]}  # type: ignore[unreachable]  # incorrect narrowing
     elif isinstance(select, int | sc.Variable) or isinstance(select, slice):
         check_1d()
         return {dims[0]: select}
@@ -86,12 +98,14 @@ def check_1d():
     return select.copy()
 
 
-def to_plain_index(dims: list[str], select: ScippIndex) -> int | slice | tuple:
+def to_plain_index(
+    dims: Sequence[str], select: ScippIndex
+) -> int | slice | tuple[int | slice, ...]:
     """
     Given a valid "scipp" index 'select', return an equivalent plain numpy-style index.
     """
-    select = _to_canonical_select(dims, select)
-    index = [slice(None)] * len(dims)
+    select = to_canonical_select(dims, select)
+    index: list[int | slice] = [slice(None)] * len(dims)
     for key, sel in select.items():
         if key not in dims:
             raise sc.DimensionError(
@@ -104,8 +118,8 @@ def to_plain_index(dims: list[str], select: ScippIndex) -> int | slice | tuple:
 
 
 def to_child_select(
-    dims: list[str],
-    child_dims: list[str],
+    dims: Sequence[str],
+    child_dims: Sequence[str],
     select: ScippIndex,
     bin_edge_dim: str | None = None,
 ) -> ScippIndex:
@@ -115,7 +129,7 @@ def to_child_select(
 
     This removes any selections that apply to the parent but not the child.
     """
-    select = _to_canonical_select(dims, select)
+    select = to_canonical_select(dims, select)
     for d in dims:
         if d not in child_dims and d in select:
             del select[d]

src/scippnexus/_load.py

@@ -2,7 +2,9 @@
 # Copyright (c) 2023 Scipp contributors (https://github.com/scipp)
 import contextlib
 import io
+from contextlib import AbstractContextManager
 from os import PathLike
+from typing import Any
 
 import h5py as h5
 import scipp as sc
@@ -23,7 +25,11 @@ def load(
     root: str | None = None,
     select: ScippIndex = (),
     definitions: Definitions | DefaultDefinitionsType = DefaultDefinitions,
-) -> sc.DataGroup | sc.DataArray | sc.Dataset:
+) -> (
+    sc.DataGroup[sc.DataGroup[Any] | sc.DataArray | sc.Dataset]
+    | sc.DataArray
+    | sc.Dataset
+):
     """Load a NeXus file.
 
     This function is a shorthand for opening a file manually.
@@ -98,7 +104,7 @@ def load(
 def _open(
     filename: str | PathLike[str] | io.BytesIO | h5.Group | Group,
     definitions: Definitions | DefaultDefinitionsType = DefaultDefinitions,
-):
+) -> AbstractContextManager[Group | File]:
     if isinstance(filename, h5.Group):
         return contextlib.nullcontext(
             Group(

src/scippnexus/_hdf5_nexus.py src/scippnexus/_string.pysrc/scippnexus/_hdf5_nexus.py renamed to src/scippnexus/_string.py

@@ -2,13 +2,11 @@
 # Copyright (c) 2023 Scipp contributors (https://github.com/scipp)
 # @author Matthew Jones
 import warnings
-from typing import Any
 
 import h5py
-import numpy as np
 
 
-def _cset_to_encoding(cset: int) -> str:
+def cset_to_encoding(cset: int) -> str:
     """
     Converts a HDF5 cset into a python encoding. Allowed values for cset are
     h5py.h5t.CSET_ASCII and h5py.h5t.CSET_UTF8.
@@ -32,9 +30,9 @@ def _cset_to_encoding(cset: int) -> str:
         )
 
 
-def _warn_latin1_decode(obj, decoded, error):
+def warn_latin1_decode(obj: object, decoded: str, error: Exception) -> None:
     warnings.warn(
-        f"Encoding for bytes '{obj}' declared as ascii, "
+        f"Encoding for bytes '{obj!r}' declared as ascii, "
         f"but contains characters in extended ascii range. Assuming "
         f"extended ASCII (latin-1), but this behavior is not "
         f"specified by the HDF5 or nexus standards and may therefore "
@@ -44,7 +42,7 @@ def _warn_latin1_decode(obj, decoded, error):
     )
 
 
-def _ensure_str(str_or_bytes: str | bytes, encoding: str) -> str:
+def ensure_str(str_or_bytes: str | bytes, encoding: str) -> str:
     """
     See https://docs.h5py.org/en/stable/strings.html for justification about some of
     the operations performed in this method. In particular, variable-length strings
@@ -66,21 +64,7 @@ def _ensure_str(str_or_bytes: str | bytes, encoding: str) -> str:
             return str(str_or_bytes, encoding="ascii")
         except UnicodeDecodeError as e:
             decoded = str(str_or_bytes, encoding="latin-1")
-            _warn_latin1_decode(str_or_bytes, decoded, str(e))
+            warn_latin1_decode(str_or_bytes, decoded, e)
             return decoded
     else:
         return str(str_or_bytes, encoding)
-
-
-_map_to_supported_type = {
-    'int8': np.int32,
-    'int16': np.int32,
-    'uint8': np.int32,
-    'uint16': np.int32,
-    'uint32': np.int32,
-    'uint64': np.int64,
-}
-
-
-def _ensure_supported_int_type(dataset_type: Any):
-    return _map_to_supported_type.get(dataset_type, dataset_type)

src/scippnexus/attrs.py

@@ -4,11 +4,13 @@
 from collections.abc import Iterator, Mapping
 from typing import Any
 
-from ._hdf5_nexus import _cset_to_encoding, _ensure_str
+import h5py as h5
 
+from ._string import cset_to_encoding, ensure_str
 
-class Attrs(Mapping):
-    def __init__(self, attrs: Mapping):
+
+class Attrs(Mapping[str, Any]):
+    def __init__(self, attrs: h5.AttributeManager) -> None:
         self._base_attrs = attrs
         self._attrs = dict(attrs)
 
@@ -17,7 +19,7 @@ def __getitem__(self, name: str) -> Any:
         # Is this check for string attributes sufficient? Is there a better way?
         if isinstance(attr, str | bytes):
             cset = self._base_attrs.get_id(name.encode("utf-8")).get_type().get_cset()
-            return _ensure_str(attr, _cset_to_encoding(cset))
+            return ensure_str(attr, cset_to_encoding(cset))
         return attr
 
     def __iter__(self) -> Iterator[str]: