Add EIT frequency blocks and multiple injection patterns

README.md

@@ -70,3 +70,53 @@ Email: jacob.thoenes@uni-rostock.de
 > Thank you!
 
 ---
+
+## EIT frequency blocks and multiple injection patterns
+
+The existing scalar frequency and `exc_freq_max` / `n_freq` / `freq_scale`
+sweep configuration remain supported. To configure several sweeps or discrete
+frequencies in one measurement, pass ordered `EitFrequencyBlock` objects:
+
+```python
+from sciopy import EitFrequencyBlock, EitMeasurementSetup
+
+setup = EitMeasurementSetup(
+    burst_count=2,
+    n_el=16,
+    exc_freq=[
+        EitFrequencyBlock(1_000, 10_000, 10, "lin"),
+        EitFrequencyBlock(10_000, 100_000, 20, "log"),
+    ],
+    framerate=2,
+    amplitude=0.001,
+    inj_skip=[0, 1],
+    gain=1,
+    adc_range=1,
+)
+# With a connected EIT device:
+# device.SetMeasurementSetup(setup)
+# data = device.StartStopMeasurement(return_as="pot_mat")
+```
+
+Do not combine block lists with the legacy sweep fields. Each injection skip
+adds `n_el` excitation settings in the order given. Skip values must be distinct.
+`SetMeasurementSetup` resets the device configuration before appending blocks
+and injection pairs. Direct frequency-update calls append to the device stack;
+use a complete setup to replace an earlier configuration.
+
+For multiple frequencies, `pot_mat` has shape
+`(frames, excitation settings, frequencies, channels)`. The example above
+returns `(2, 32, 30, 16)` for two complete frames. A single frequency keeps the
+existing `(frames, excitation settings, channels)` shape. Frame
+`frequency_stgs` contains the configured frequencies **in Hz**, computed from
+the requested blocks; these are not device readback or independently measured
+frequencies. The wire frequency-row identifier is an index into that list.
+
+Software tests verify command encoding and synthetic frame parsing. Physical
+device validation is still required for each supported model and firmware.
+
+Measurement reads tolerate transient empty USB reads and stop after the requested
+number of complete frames, including saved frames deleted from RAM. The default
+idle timeout is five seconds; slower measurements may require a larger
+`idle_timeout_seconds` when calling `MessageParser.read_usb_till_timeout`.
+Command-response reads retain their immediate empty-read timeout.

sciopy/EIT_16_32_64_128.py

@@ -11,7 +11,7 @@
 import numpy as np
 from pyftdi.ftdi import Ftdi
 
-from .sciopy_dataclasses import EitMeasurementSetup
+from .sciopy_dataclasses import EitFrequencyBlock, EitMeasurementSetup
 from .usb_message_parser import (
     MessageParser,
     make_eitframes_hex,
@@ -313,60 +313,69 @@ def update_ExcitationFrequencies(
           parsed with the right number of frequency points.
         - Sends the configuration command to the connected device.
         """
-        scale_options = {"lin": 0x00, "log": 0x01}
-        if f_scale not in scale_options:
-            raise ValueError(f"Unknown f_scale {f_scale!r}; expected 'lin' or 'log'.")
-        if f_count < 1:
-            raise ValueError("f_count must be >= 1.")
-        if f_max is None:
-            f_max = f_min
-        if f_count > 1 and f_max <= f_min:
-            raise ValueError("f_max must be greater than f_min when f_count > 1.")
-
+        block = EitFrequencyBlock(
+            f_min, f_min if f_max is None else f_max, f_count, f_scale
+        )
+        block.validate()
+        self._write_frequency_blocks([block])
         if self.setup is not None:
             self.setup.exc_freq = f_min
-            self.setup.exc_freq_max = f_max
+            self.setup.exc_freq_max = block.f_max
             self.setup.n_freq = f_count
-            self.setup.freq_scale = f_scale
+            self.setup.freq_scale = block.scale
             if self.cMessageParser is not None:
-                # Re-sync the parser's frame layout (iNumFreqSettings etc.)
-                # with the new sweep configuration.
                 self.cMessageParser.set_measurement_setup(self.setup)
 
-        # Set frequencies:
-        # [CT] 0C 04 [fmin] [fmax] [fcount] [fscale] [CT]
-        self.print_msg = True
-        self.write_command_string(
-            bytearray(
-                list(
-                    np.concatenate(
-                        [
-                            [176, 12, 4],
-                            clTbt_sp(f_min),
-                            clTbt_sp(f_max),
-                            clTbt_u16(f_count),
-                            [scale_options[f_scale]],
-                            [176],
-                        ]
-                    )
+    def _write_frequency_blocks(self, blocks):
+        # Build every command before writing, so invalid input cannot partially
+        # configure the device. Each command appends one frequency-stack block.
+        commands = []
+        for block in blocks:
+            block.validate()
+            commands.append(
+                bytearray(
+                    [0xB0, 0x0C, 0x04]
+                    + clTbt_sp(block.f_min)
+                    + clTbt_sp(block.f_max)
+                    + clTbt_u16(int(block.f_count))
+                    + [0 if block.scale == "lin" else 1, 0xB0]
                 )
             )
-        )
-        self.print_msg = False
+        previous_print_msg = self.print_msg
+        self.print_msg = True
+        try:
+            for command in commands:
+                self.write_command_string(command)
+        finally:
+            self.print_msg = previous_print_msg
 
     def update_ExcitationFrequency(self, exc_freq):
-        """
-        Sets a single excitation frequency (no sweep).
-
-        Kept for backward compatibility; equivalent to
-        ``update_ExcitationFrequencies(f_min=exc_freq, f_count=1)``.
+        """Configure a scalar frequency or append an ordered list of blocks.
 
-        Parameters
-        ----------
-        exc_freq : int or float
-            frequency to be set from 100 Hz to 1 MHz
+        Scalar calls retain the existing single-frequency API. Block lists
+        synchronize the parser with their combined frequency count. Use
+        SetMeasurementSetup to reset the device's frequency stack first.
         """
-        self.update_ExcitationFrequencies(f_min=exc_freq, f_count=1)
+        if not isinstance(exc_freq, (list, tuple)):
+            self.update_ExcitationFrequencies(f_min=exc_freq, f_count=1)
+            return
+        blocks = list(exc_freq)
+        if not blocks:
+            raise ValueError("The frequency block list cannot be empty.")
+        for block in blocks:
+            if not isinstance(block, EitFrequencyBlock):
+                raise TypeError(
+                    "Every frequency block must be an EitFrequencyBlock instance."
+                )
+            block.validate()
+        self._write_frequency_blocks(blocks)
+        if self.setup is not None:
+            self.setup.exc_freq = blocks
+            self.setup.exc_freq_max = None
+            self.setup.n_freq = 1
+            self.setup.freq_scale = "lin"
+            if self.cMessageParser is not None:
+                self.cMessageParser.set_measurement_setup(self.setup)
 
     def SetMeasurementSetup(self, setup: EitMeasurementSetup):
         """
@@ -403,6 +412,8 @@ def SetMeasurementSetup(self, setup: EitMeasurementSetup):
                 "EIT_16_32_64_128 initialization."
             )
 
+        setup.get_frequency_blocks()
+        injection_skips = setup.get_injection_skips()
         self.setup = setup
         self.cMessageParser.set_measurement_setup(self.setup)
         self.print_msg = False
@@ -463,22 +474,24 @@ def SetMeasurementSetup(self, setup: EitMeasurementSetup):
                 list(np.concatenate([[176, 5, 3], clTbt_sp(setup.framerate), [176]]))
             )
         )
-        # Set excitation frequencies (single frequency, or a full sweep when
-        # setup.n_freq > 1):
-        self.update_ExcitationFrequencies(
-            f_min=setup.exc_freq,
-            f_max=(
-                setup.exc_freq_max if setup.exc_freq_max is not None else setup.exc_freq
-            ),
-            f_count=setup.n_freq,
-            f_scale=setup.freq_scale,
-        )
+        if isinstance(setup.exc_freq, (list, tuple)):
+            self.update_ExcitationFrequency(setup.exc_freq)
+        else:
+            self.update_ExcitationFrequencies(
+                f_min=setup.exc_freq,
+                f_max=setup.exc_freq_max,
+                f_count=setup.n_freq,
+                f_scale=setup.freq_scale,
+            )
 
-        # Set injection config
+        # Append a complete set of injection pairs for each skip pattern.
         el_inj = np.arange(1, setup.n_el + 1)
-        el_gnd = np.roll(el_inj, -(setup.inj_skip + 1))
-        for v_el, g_el in zip(el_inj, el_gnd):
-            self.write_command_string(bytearray([0xB0, 0x03, 0x06, v_el, g_el, 0xB0]))
+        for skip in injection_skips:
+            el_gnd = np.roll(el_inj, -(skip + 1))
+            for v_el, g_el in zip(el_inj, el_gnd):
+                self.write_command_string(
+                    bytearray([0xB0, 0x03, 0x06, int(v_el), int(g_el), 0xB0])
+                )
 
         self.print_msg = True
         # Set output configuration - enable all

sciopy/__init__.py

@@ -6,11 +6,13 @@
 
 from .EIT_16_32_64_128 import EIT_16_32_64_128, EitMeasurementSetup
 from .ISX_3 import ISX_3, EisMeasurementSetup
+from .sciopy_dataclasses import EitFrequencyBlock
 
 __all__ = [
     "available_serial_ports",
     "EIT_16_32_64_128",
     "EitMeasurementSetup",
+    "EitFrequencyBlock",
     "ISX_3",
     "EisMeasurementSetup",
 ]

sciopy/sciopy_dataclasses.py

@@ -1,10 +1,51 @@
 """Dataclasses for sciopy package."""
 
 from dataclasses import dataclass
+from math import isfinite
+from numbers import Integral, Real
+import numpy as np
 from typing import List, Tuple, Union
 import numpy.typing as npt
 
 
+@dataclass
+class EitFrequencyBlock:
+    """One excitation block, with frequencies in Hz and a lin/log distribution."""
+
+    f_min: Union[int, float]
+    f_max: Union[int, float]
+    f_count: int = 1
+    f_type: str = "lin"
+
+    def validate(self):
+        """Validate before sending commands or allocating parser buffers."""
+        for value in (self.f_min, self.f_max):
+            if isinstance(value, bool) or not isinstance(value, Real):
+                raise TypeError("Frequency endpoints must be real numbers.")
+            if not isfinite(value) or value <= 0:
+                raise ValueError("Frequency endpoints must be finite and positive.")
+        if isinstance(self.f_count, bool) or not isinstance(self.f_count, Integral):
+            raise TypeError("f_count must be an integer.")
+        if not 1 <= self.f_count <= 0xFFFF:
+            raise ValueError("f_count must fit in a nonzero unsigned 16-bit integer.")
+        if self.f_max < self.f_min or (self.f_count > 1 and self.f_max == self.f_min):
+            raise ValueError("f_max must be greater than f_min when f_count > 1.")
+        if not isinstance(self.f_type, str) or self.f_type.lower() not in {
+            "lin",
+            "linear",
+            "log",
+            "logarithmic",
+        }:
+            raise ValueError("f_type must be 'lin' or 'log'.")
+
+    @property
+    def scale(self):
+        """Canonical spelling used by the wire protocol."""
+        return {"lin": "lin", "linear": "lin", "log": "log", "logarithmic": "log"}[
+            self.f_type.lower()
+        ]
+
+
 @dataclass
 class EitMeasurementSetup:
     """
@@ -13,11 +54,13 @@ class EitMeasurementSetup:
     Attributes:
         burst_count (int): Number of bursts per measurement cycle.
         n_el (int): Number of electrodes used in the measurement.
-        exc_freq (int or float): Excitation frequency in Hz. Acts as the start
+        exc_freq (int, float or list of EitFrequencyBlock): Frequency in Hz,
+            or ordered sweep blocks. A scalar acts as the start
             (minimum) frequency of a sweep when ``n_freq`` > 1.
         framerate (int or float): Frame rate of the measurement in Hz.
         amplitude (int or float): Amplitude of the excitation signal.
-        inj_skip (int or list): Electrode(s) to skip during current injection.
+        inj_skip (int or list): One or several distinct injection skip patterns.
+            Each pattern adds n_el excitation settings in configured order.
         gain (int): Amplifier gain setting.
         adc_range (int): Analog-to-digital converter range setting.
         mea_mode (str): Measurement mode, see `update_measurement_mode`.
@@ -36,7 +79,7 @@ class EitMeasurementSetup:
 
     burst_count: int
     n_el: int
-    exc_freq: Union[int, float]
+    exc_freq: Union[int, float, List[EitFrequencyBlock]]
     framerate: Union[int, float]
     amplitude: Union[int, float]
     inj_skip: Union[int, list]
@@ -48,6 +91,55 @@ class EitMeasurementSetup:
     n_freq: int = 1
     freq_scale: str = "lin"
 
+    def get_frequency_blocks(self):
+        """Resolve either the legacy sweep fields or an explicit block list."""
+        if isinstance(self.exc_freq, (list, tuple)):
+            if (
+                self.exc_freq_max is not None
+                or self.n_freq != 1
+                or self.freq_scale != "lin"
+            ):
+                raise ValueError(
+                    "Do not combine frequency blocks with legacy sweep fields."
+                )
+            blocks = list(self.exc_freq)
+        else:
+            blocks = [
+                EitFrequencyBlock(
+                    self.exc_freq,
+                    self.exc_freq if self.exc_freq_max is None else self.exc_freq_max,
+                    self.n_freq,
+                    self.freq_scale,
+                )
+            ]
+        if not blocks:
+            raise ValueError("The frequency block list cannot be empty.")
+        for block in blocks:
+            if not isinstance(block, EitFrequencyBlock):
+                raise TypeError(
+                    "Every frequency block must be an EitFrequencyBlock instance."
+                )
+            block.validate()
+        return blocks
+
+    def get_injection_skips(self):
+        """Return validated skip patterns in their configured order."""
+        skips = (
+            list(self.inj_skip)
+            if isinstance(self.inj_skip, (list, tuple, np.ndarray))
+            else [self.inj_skip]
+        )
+        if not skips:
+            raise ValueError("inj_skip cannot be empty.")
+        for skip in skips:
+            if isinstance(skip, bool) or not isinstance(skip, Integral):
+                raise TypeError("Every inj_skip value must be an integer.")
+            if not 0 <= skip < self.n_el - 1:
+                raise ValueError("inj_skip must satisfy 0 <= inj_skip < n_el - 1.")
+        if len(set(skips)) != len(skips):
+            raise ValueError("inj_skip values must be unique.")
+        return [int(skip) for skip in skips]
+
 
 @dataclass
 class EisMeasurementSetup: