Add tests for collections and tools; improve numpy2 support

.github/workflows/test.yml

@@ -22,7 +22,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ['3.9', '3.10', '3.11', '3.12', ]
+        python-version: ['3.9', '3.10', '3.11', '3.12', '3.13', ]
 
     steps:
     - uses: actions/checkout@v3

pyproject.toml

@@ -31,6 +31,7 @@ classifiers = [
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
     "Development Status :: 5 - Production/Stable",
     "Intended Audience :: Science/Research",
     "License :: OSI Approved :: BSD License",
@@ -46,7 +47,7 @@ dependencies = [
     "lmfit >= 1.0.0",
     "matplotlib >= 3.1.2",
     "mrcfile >= 1.1.2",
-    "numpy >= 1.17.3, <2",
+    "numpy >= 1.17.3",
     "pandas >= 1.0.0",
     "pillow >= 7.0.0",
     "pywinauto >= 0.6.8; sys_platform == 'windows'",

requirements.txt

@@ -5,7 +5,7 @@ ipython >= 7.11.1
 lmfit >= 1.0.0
 matplotlib >= 3.1.2
 mrcfile >= 1.1.2
-numpy >= 1.17.3, <2
+numpy >= 1.17.3
 pandas >= 1.0.0
 pillow >= 7.0.0
 pywinauto >= 0.6.8; sys_platform == 'windows'

scripts/process_dm.py

@@ -8,6 +8,7 @@
 from skimage.exposure import rescale_intensity
 
 from instamatic.processing.ImgConversionDM import ImgConversionDM as ImgConversion
+from instamatic.tools import relativistic_wavelength
 
 # Script to process cRED data collecting using the DigitalMicrograph script `insteaDMatic`
 # https://github.com/instamatic-dev/InsteaDMatic
@@ -24,21 +25,6 @@
 # all `cred_log.txt` files in the subdirectories, and iterate over those.
 
 
-def relativistic_wavelength(voltage: float = 200):
-    """Calculate the relativistic wavelength of electrons Voltage in kV Return
-    wavelength in Angstrom."""
-    voltage *= 1000  # -> V
-
-    h = 6.626070150e-34  # planck constant J.s
-    m = 9.10938356e-31  # electron rest mass kg
-    e = 1.6021766208e-19  # elementary charge C
-    c = 299792458  # speed of light m/s
-
-    wl = h / (2 * m * voltage * e * (1 + (e * voltage) / (2 * m * c**2))) ** 0.5
-
-    return round(wl * 1e10, 6)  # m -> Angstrom
-
-
 def img_convert(credlog, tiff_path='tiff2', mrc_path='RED', smv_path='SMV'):
     credlog = Path(credlog)
     drc = credlog.parent
@@ -90,7 +76,7 @@ def img_convert(credlog, tiff_path='tiff2', mrc_path='RED', smv_path='SMV'):
             if line.startswith('Resolution:'):
                 resolution = line.split()[-1]
 
-    wavelength = relativistic_wavelength(high_tension)
+    wavelength = relativistic_wavelength(high_tension * 1000)
 
     # convert from um to mm
     physical_pixelsize = physical_pixelsize[0] / 1000

src/instamatic/_collections.py

@@ -1,11 +1,10 @@
 from __future__ import annotations
 
-import contextlib
 import logging
 import string
-import time
 from collections import UserDict
-from typing import Any, Callable
+from dataclasses import dataclass
+from typing import Any
 
 
 class NoOverwriteDict(UserDict):
@@ -18,6 +17,8 @@ def __setitem__(self, key: Any, value: Any) -> None:
 
 
 class NullLogger(logging.Logger):
+    """A logger mock that ignores all logging, to be used in headless mode."""
+
     def __init__(self, name='null'):
         super().__init__(name)
         self.addHandler(logging.NullHandler())
@@ -27,24 +28,28 @@ def __init__(self, name='null'):
 class PartialFormatter(string.Formatter):
     """`str.format` alternative, allows for partial replacement of {fields}"""
 
+    @dataclass(frozen=True)
+    class Missing:
+        name: str
+
     def __init__(self, missing: str = '{{{}}}') -> None:
         super().__init__()
         self.missing: str = missing  # used instead of missing values
 
     def get_field(self, field_name: str, args, kwargs) -> tuple[Any, str]:
         """When field can't be found, return placeholder text instead."""
         try:
-            obj, used_key = super().get_field(field_name, args, kwargs)
-            return obj, used_key
+            return super().get_field(field_name, args, kwargs)
         except (KeyError, AttributeError, IndexError, TypeError):
-            return self.missing.format(field_name), field_name
+            return PartialFormatter.Missing(field_name), field_name
 
     def format_field(self, value: Any, format_spec: str) -> str:
         """If the field was not found, format placeholder as string instead."""
-        try:
-            return super().format_field(value, format_spec)
-        except (ValueError, TypeError):
-            return str(value)
+        if isinstance(value, PartialFormatter.Missing):
+            if format_spec:
+                return self.missing.format(f'{value.name}:{format_spec}')
+            return self.missing.format(f'{value.name}')
+        return super().format_field(value, format_spec)
 
 
 partial_formatter = PartialFormatter()

src/instamatic/calibrate/calibrate_stage_rotation.py

@@ -163,7 +163,7 @@ def to_file(self, outdir: Optional[str] = None) -> None:
             outdir = calibration_drc
         yaml_path = Path(outdir) / CALIB_STAGE_ROTATION
         with open(yaml_path, 'w') as yaml_file:
-            yaml.safe_dump(asdict(self), yaml_file)  # noqa: correct type
+            yaml.safe_dump(asdict(self), yaml_file)  # type: ignore[arg-type]
         log(f'{self} saved to {yaml_path}.')
 
     def plot(self, sst: Optional[list[SpanSpeedTime]] = None) -> None:

src/instamatic/camera/gatansocket3.py

@@ -86,6 +86,17 @@
 sArgsBuffer = np.zeros(ARGS_BUFFER_SIZE, dtype=np.byte)
 
 
+def string_to_longarray(string: str, *, dtype: np.dtype = np.int_) -> np.ndarray:
+    """Convert the string to a 1D np array of dtype (default np.int_ - C long)
+    with numpy2-save padding to ensure length is a multiple of dtype.itemsize.
+    """
+    s_bytes = string.encode('utf-8')
+    dtype_size = np.dtype(dtype).itemsize
+    if extra := len(s_bytes) % dtype_size:
+        s_bytes += b'\0' * (dtype_size - extra)
+    return np.frombuffer(s_bytes, dtype=dtype)
+
+
 class Message:
     """Information packet to send and receive on the socket.
 
@@ -335,14 +346,7 @@ def SetK2Parameters(
         funcCode = enum_gs['GS_SetK2Parameters']
 
         self.save_frames = saveFrames
-
-        # filter name
-        filt_str = filt + '\0'
-        extra = len(filt_str) % 4
-        if extra:
-            npad = 4 - extra
-            filt_str = filt_str + npad * '\0'
-        longarray = np.frombuffer(filt_str.encode(), dtype=np.int_)
+        longarray = string_to_longarray(filt + '\0', dtype=np.int_)  # filter name
 
         longs = [
             funcCode,
@@ -397,12 +401,7 @@ def SetupFileSaving(
             longs = [enum_gs['GS_SetupFileSaving'], rotationFlip]
             dbls = [pixelSize]
         bools = [filePerImage]
-        names_str = dirname + '\0' + rootname + '\0'
-        extra = len(names_str) % 4
-        if extra:
-            npad = 4 - extra
-            names_str = names_str + npad * '\0'
-        longarray = np.frombuffer(names_str.encode(), dtype=np.int_)
+        longarray = string_to_longarray(dirname + '\0' + rootname + '\0', dtype=np.int_)
         message_send = Message(
             longargs=longs, boolargs=bools, dblargs=dbls, longarray=longarray
         )
@@ -664,24 +663,18 @@ def ExecuteScript(
         select_camera=0,
         recv_longargs_init=(0,),
         recv_dblargs_init=(0.0,),
-        recv_longarray_init=[],
+        recv_longarray_init=None,
     ):
+        """Send the command string as a 1D longarray of np.int_ dtype."""
         funcCode = enum_gs['GS_ExecuteScript']
-        cmd_str = command_line + '\0'
-        extra = len(cmd_str) % 4
-        if extra:
-            npad = 4 - extra
-            cmd_str = cmd_str + (npad) * '\0'
-        # send the command string as 1D longarray
-        longarray = np.frombuffer(cmd_str.encode(), dtype=np.int_)
-        # print(longaray)
+        longarray = string_to_longarray(command_line + '\0', dtype=np.int_)
         message_send = Message(
             longargs=(funcCode,), boolargs=(select_camera,), longarray=longarray
         )
         message_recv = Message(
             longargs=recv_longargs_init,
             dblargs=recv_dblargs_init,
-            longarray=recv_longarray_init,
+            longarray=[] if recv_longarray_init is None else recv_longarray_init,
         )
         self.ExchangeMessages(message_send, message_recv)
         return message_recv

src/instamatic/gui/fast_adt_frame.py

@@ -70,8 +70,8 @@ def as_dict(self):
 class ExperimentalFastADT(LabelFrame):
     """GUI panel to perform selected FastADT-style (c)RED & PED experiments."""
 
-    def __init__(self, parent):  # noqa: parent.__init__ is called
-        LabelFrame.__init__(self, parent, text='Experiment with a priori tracking options')
+    def __init__(self, parent):
+        super().__init__(parent, text='Experiment with a priori tracking options')
         self.parent = parent
         self.var = ExperimentalFastADTVariables()
         self.q: Optional[Queue] = None

src/instamatic/tools.py

@@ -1,14 +1,11 @@
 from __future__ import annotations
 
-import glob
-import os
 import sys
 from pathlib import Path
-from typing import Tuple
+from typing import Iterator
 
 import numpy as np
 from scipy import interpolate, ndimage
-from skimage import exposure
 from skimage.measure import regionprops
 
 
@@ -71,9 +68,13 @@ def to_xds_untrusted_area(kind: str, coords: list) -> str:
         raise ValueError('Only quadrilaterals are supported for now')
 
 
-def find_subranges(lst: list) -> Tuple[int, int]:
+def find_subranges(lst: list[int]) -> Iterator[tuple[int, int]]:
     """Takes a range of sequential numbers (possibly with gaps) and splits them
-    in sequential sub-ranges defined by the minimum and maximum value."""
+    in sequential sub-ranges defined by the minimum and maximum value.
+
+    Example:
+        [1,2,3,7,8,10] --> (1,3), (7,8), (10,10)
+    """
     from itertools import groupby
     from operator import itemgetter
 
@@ -274,7 +275,7 @@ def get_acquisition_time(
 
 
 def relativistic_wavelength(voltage: float = 200_000) -> float:
-    """Calculate the relativistic wavelength of electrons from the accelarating
+    """Calculate the relativistic wavelength of electrons from the accelerating
     voltage.
 
     Input: Voltage in V

tests/test_collections.py

@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+import logging
+from contextlib import nullcontext
+from dataclasses import dataclass, field
+from typing import Any, Optional, Type
+
+import pytest
+
+import instamatic._collections as ic
+from tests.utils import InstanceAutoTracker
+
+
+def test_no_overwrite_dict() -> None:
+    """Should work as normal dict unless key exists, in which case raises."""
+    nod = ic.NoOverwriteDict({1: 2})
+    nod.update({3: 4})
+    nod[5] = 6
+    del nod[1]
+    nod[1] = 6
+    assert nod == {1: 6, 3: 4, 5: 6}
+    with pytest.raises(KeyError):
+        nod[1] = 2
+    with pytest.raises(KeyError):
+        nod.update({3: 4})
+
+
+def test_null_logger(caplog) -> None:
+    """NullLogger should void and not propagate messages to root logger."""
+
+    messages = []
+    handler = logging.StreamHandler()
+    handler.emit = lambda record: messages.append(record.getMessage())
+    null_logger = ic.NullLogger()
+    root_logger = logging.getLogger()
+    root_logger.addHandler(handler)
+
+    with caplog.at_level(logging.DEBUG):
+        null_logger.debug('debug message that should be ignored')
+        null_logger.info('info message that should be ignored')
+        null_logger.warning('warning message that should be ignored')
+        null_logger.error('error message that should be ignored')
+        null_logger.critical('critical message that should be ignored')
+
+    # Nothing should have been captured by pytest's caplog
+    root_logger.removeHandler(handler)
+    assert caplog.records == []
+    assert caplog.text == ''
+    assert messages == []
+
+
+@dataclass
+class PartialFormatterTestCase(InstanceAutoTracker):
+    template: str = '{s} & {f:06.2f}'
+    args: list[Any] = field(default_factory=list)
+    kwargs: dict[str, Any] = field(default_factory=dict)
+    returns: str = ''
+    raises: Optional[Type[Exception]] = None
+
+
+PartialFormatterTestCase(returns='{s} & {f:06.2f}')
+PartialFormatterTestCase(kwargs={'s': 'Text'}, returns='Text & {f:06.2f}')
+PartialFormatterTestCase(kwargs={'f': 3.1415}, returns='{s} & 003.14')
+PartialFormatterTestCase(kwargs={'x': 'test'}, returns='{s} & {f:06.2f}')
+PartialFormatterTestCase(kwargs={'f': 'Text'}, raises=ValueError)
+PartialFormatterTestCase(template='{0}{1}', args=[5], returns='5{1}')
+PartialFormatterTestCase(template='{0}{1}', args=[5, 6], returns='56')
+PartialFormatterTestCase(template='{0}{1}', args=[5, 6, 7], returns='56')
+
+
+@pytest.mark.parametrize('test_case', PartialFormatterTestCase.INSTANCES)
+def test_partial_formatter(test_case) -> None:
+    """Should replace only some {words}, but still fail if format is wrong."""
+    c = test_case
+    with pytest.raises(r) if (r := c.raises) else nullcontext():
+        assert ic.partial_formatter.format(c.template, *c.args, **c.kwargs) == c.returns