From 9f983f0531839d58517ce87db4084478dc469414 Mon Sep 17 00:00:00 2001 From: dokudoo Date: Thu, 8 Oct 2026 17:57:31 +0200 Subject: [PATCH] Add EIT frequency blocks and multiple injection patterns --- README.md | 50 +++++ sciopy/EIT_16_32_64_128.py | 127 ++++++----- sciopy/__init__.py | 2 + sciopy/sciopy_dataclasses.py | 98 ++++++++- sciopy/usb_message_parser.py | 75 ++++--- tests/test_eit_frequency_sweep.py | 243 +++++++++++++++++++++ tests/test_usb_message_parser.py | 340 +++++++++++++++++++++++++++++- 7 files changed, 842 insertions(+), 93 deletions(-) create mode 100644 tests/test_eit_frequency_sweep.py diff --git a/README.md b/README.md index 6e8afdf..0a4f1b8 100644 --- a/README.md +++ b/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. diff --git a/sciopy/EIT_16_32_64_128.py b/sciopy/EIT_16_32_64_128.py index bf20496..fd348a5 100644 --- a/sciopy/EIT_16_32_64_128.py +++ b/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 diff --git a/sciopy/__init__.py b/sciopy/__init__.py index fa99657..5af3e5c 100644 --- a/sciopy/__init__.py +++ b/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", ] diff --git a/sciopy/sciopy_dataclasses.py b/sciopy/sciopy_dataclasses.py index 7958b14..afe8e1a 100644 --- a/sciopy/sciopy_dataclasses.py +++ b/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: diff --git a/sciopy/usb_message_parser.py b/sciopy/usb_message_parser.py index 1037c07..f86e967 100644 --- a/sciopy/usb_message_parser.py +++ b/sciopy/usb_message_parser.py @@ -147,6 +147,7 @@ def __init__(self, device, eitsetup=None, devicetype="FS"): # General setup self.bPrintMessages = False self.iNPZSaveIndex = 1 + self._completed_frames = 0 self.iSaveCounter = 0 # Unused self.ppcData = [] self.iInjIndex = 0 @@ -187,8 +188,10 @@ def set_measurement_setup(self, setup: EitMeasurementSetup): self.setup = setup if setup is not None: self.iMaxChannelGroups = setup.n_el // 16 - self.iNumExcitationSettings = setup.n_el # todo should be independently set - self.iNumFreqSettings = getattr(setup, "n_freq", None) or 1 + self.iNumExcitationSettings = setup.n_el * len(setup.get_injection_skips()) + self.iNumFreqSettings = sum( + block.f_count for block in setup.get_frequency_blocks() + ) self.iLenDataperFrame = ( self.iMaxChannelGroups * 16 @@ -245,18 +248,17 @@ def get_frequency_list(self) -> np.ndarray: if self.setup is None: return np.zeros((self.iNumFreqSettings,), dtype=float) - f_min = self.setup.exc_freq - n_freq = getattr(self.setup, "n_freq", None) or 1 - if n_freq <= 1: - return np.array([f_min], dtype=float) - - f_max = getattr(self.setup, "exc_freq_max", None) - if f_max is None: - f_max = f_min - f_scale = getattr(self.setup, "freq_scale", "lin") - if f_scale == "log": - return np.logspace(np.log10(f_min), np.log10(f_max), n_freq) - return np.linspace(f_min, f_max, n_freq) + frequencies = [] + for block in self.setup.get_frequency_blocks(): + if block.f_count == 1: + frequencies.append(np.array([block.f_min], dtype=float)) + elif block.scale == "log": + frequencies.append( + np.geomspace(block.f_min, block.f_max, block.f_count) + ) + else: + frequencies.append(np.linspace(block.f_min, block.f_max, block.f_count)) + return np.concatenate(frequencies) # ---------------------------------------------------------------------------------------------------------------- # def clear_out_data(self): @@ -389,21 +391,22 @@ def read_usb_till_timeout( bDeleteDataFrame: bool = False, sSavePath: str = "C/", bStartReset: bool = True, + idle_timeout_seconds: float = 5.0, ): + """Read a measurement through transient empty reads, or until idle. + + Command-response reads (bStartReset=False or no setup) retain their + immediate empty-read timeout. Measurement reads stop after the requested + number of complete frames, including frames saved and deleted from RAM. + An idle timeout can still return fewer frames than requested. """ - Reads out the USB connection until the connections times out, so for messages received + timeout. Data bytes are parsed, - sorted into full messages and then handled according to their Command Tag. Status or requested information is - displayed if wished and measured EIT data is stored, deleted or returned. - Args: - bSaveData: if data should be saved - bDeleteDataFrame: if data frame is deleted after saving data - sSavePath: Path where the data should be saved - Returns: - List of received data eit frames, no Status messages are saved - """ - if bStartReset and self.setup is not None: + if not np.isfinite(idle_timeout_seconds) or idle_timeout_seconds < 0: + raise ValueError("idle_timeout_seconds must be finite and nonnegative.") + is_measurement_read = bStartReset and self.setup is not None + if is_measurement_read: self.reset_new_data_frame() - timeout_count = 0 + start_frame_count = self._completed_frames + last_data_time = time.monotonic() while True: buffer = self.device_read() if buffer: @@ -411,13 +414,20 @@ def read_usb_till_timeout( self.interpret_message( message, bSaveData, bDeleteDataFrame, sSavePath ) - timeout_count = 0 - continue - timeout_count += 1 - if timeout_count >= 1: - # Break if we haven't received any data + last_data_time = time.monotonic() + if ( + is_measurement_read + and self.setup.burst_count > 0 + and self._completed_frames - start_frame_count + >= self.setup.burst_count + ): + break + elif not is_measurement_read: break - + elif time.monotonic() - last_data_time >= idle_timeout_seconds: + break + else: + time.sleep(0.01) return self.ppcData # ---------------------------------------------------------------------------------------------------------------- # @@ -497,6 +507,7 @@ def interpret_data_input( self.iSaveCounter += 1 if self.iSaveCounter == self.iLenDataperFrame: # Frame Full + self._completed_frames += 1 self.CurrentFrame.timestamp2 = byteintarray_to_float(message[7:11]) if bSave: save_data_frame(sSavePath, self.CurrentFrame, self.iNPZSaveIndex) diff --git a/tests/test_eit_frequency_sweep.py b/tests/test_eit_frequency_sweep.py new file mode 100644 index 0000000..f4320dc --- /dev/null +++ b/tests/test_eit_frequency_sweep.py @@ -0,0 +1,243 @@ +import struct +from unittest.mock import Mock, call + +import pytest + +from sciopy.EIT_16_32_64_128 import ( + EIT_16_32_64_128, +) +from sciopy.sciopy_dataclasses import ( + EitFrequencyBlock, + EitMeasurementSetup, +) + + +def build_frequency_command( + f_min, + f_max, + f_count, + f_type, +): + command = bytearray([0xB0, 0x0C, 0x04]) + command.extend(struct.pack(">f", f_min)) + command.extend(struct.pack(">f", f_max)) + command.extend( + f_count.to_bytes( + 2, + byteorder="big", + signed=False, + ) + ) + command.extend([f_type, 0xB0]) + return command + + +def test_scalar_frequency_remains_supported(): + device = EIT_16_32_64_128(16) + device.write_command_string = Mock() + + device.update_ExcitationFrequency(100000) + + expected = build_frequency_command( + 100000, + 100000, + 1, + 0, + ) + + device.write_command_string.assert_called_once_with(expected) + + +def test_linear_and_logarithmic_blocks(): + device = EIT_16_32_64_128(16) + device.write_command_string = Mock() + + blocks = [ + EitFrequencyBlock( + f_min=1000, + f_max=10000, + f_count=10, + f_type="lin", + ), + EitFrequencyBlock( + f_min=10000, + f_max=100000, + f_count=20, + f_type="log", + ), + ] + + device.update_ExcitationFrequency(blocks) + + expected_linear = build_frequency_command( + 1000, + 10000, + 10, + 0, + ) + expected_logarithmic = build_frequency_command( + 10000, + 100000, + 20, + 1, + ) + + assert device.write_command_string.call_count == 2 + device.write_command_string.assert_has_calls( + [ + call(expected_linear), + call(expected_logarithmic), + ] + ) + + +def test_invalid_frequency_type_is_rejected(): + device = EIT_16_32_64_128(16) + device.write_command_string = Mock() + + blocks = [ + EitFrequencyBlock( + f_min=1000, + f_max=10000, + f_count=10, + f_type="invalid", + ) + ] + + with pytest.raises( + ValueError, + match="f_type", + ): + device.update_ExcitationFrequency(blocks) + + device.write_command_string.assert_not_called() + + +def test_frequency_count_must_fit_wire_format(): + device = EIT_16_32_64_128(16) + device.write_command_string = Mock() + + blocks = [ + EitFrequencyBlock( + f_min=1000, + f_max=10000, + f_count=64, + f_type="lin", + ), + EitFrequencyBlock( + f_min=10000, + f_max=100000, + f_count=65536, + f_type="log", + ), + ] + + with pytest.raises( + ValueError, + match="unsigned 16-bit", + ): + device.update_ExcitationFrequency(blocks) + + device.write_command_string.assert_not_called() + + +def test_set_measurement_setup_forwards_frequency_blocks(): + device = EIT_16_32_64_128(16) + device.cMessageParser = Mock() + device.write_command_string = Mock() + device.update_measurement_mode = Mock() + device.update_ExcitationFrequency = Mock() + + blocks = [ + EitFrequencyBlock( + f_min=1000, + f_max=10000, + f_count=10, + f_type="lin", + ), + EitFrequencyBlock( + f_min=10000, + f_max=100000, + f_count=20, + f_type="log", + ), + ] + + setup = EitMeasurementSetup( + burst_count=2, + n_el=16, + exc_freq=blocks, + framerate=2, + amplitude=0.001, + inj_skip=0, + gain=1, + adc_range=1, + ) + + device.SetMeasurementSetup(setup) + + device.update_ExcitationFrequency.assert_called_once_with(blocks) + + +@pytest.mark.parametrize("skips", [[], [True], [-1], [15], [0, 0]]) +def test_invalid_injection_skips_send_no_commands(skips): + device = EIT_16_32_64_128(16) + device.cMessageParser = Mock() + device.write_command_string = Mock() + setup = EitMeasurementSetup(1, 16, 1000, 2, 0.001, skips, 1, 1) + with pytest.raises((ValueError, TypeError)): + device.SetMeasurementSetup(setup) + device.write_command_string.assert_not_called() + device.cMessageParser.set_measurement_setup.assert_not_called() + + +def test_multiple_skip_patterns_send_all_injection_pairs(): + device = EIT_16_32_64_128(16) + device.cMessageParser = Mock() + device.write_command_string = Mock() + setup = EitMeasurementSetup(1, 16, 1000, 2, 0.001, [0, 2], 1, 1) + device.SetMeasurementSetup(setup) + pairs = [ + list(item.args[0]) + for item in device.write_command_string.call_args_list + if item.args[0][:3] == bytearray([0xB0, 3, 6]) + ] + expected = [ + [0xB0, 3, 6, out, ((out + skip) % 16) + 1, 0xB0] + for skip in (0, 2) + for out in range(1, 17) + ] + assert pairs == expected + + +def test_direct_block_update_synchronizes_parser(): + from sciopy.usb_message_parser import MessageParser + + device = EIT_16_32_64_128(16) + device.setup = EitMeasurementSetup(1, 16, 1000, 2, 0.001, 0, 1, 1) + device.cMessageParser = MessageParser(None, device.setup) + device.write_command_string = Mock() + device.update_ExcitationFrequency([EitFrequencyBlock(1000, 10000, 3)]) + assert device.cMessageParser.iNumFreqSettings == 3 + assert len(device.cMessageParser.CurrentFrame.ppcData) == 16 * 3 * 16 + + +@pytest.mark.parametrize("value", [float("nan"), float("inf"), 0, -1, True]) +def test_invalid_frequency_endpoints_send_no_commands(value): + device = EIT_16_32_64_128(16) + device.write_command_string = Mock() + with pytest.raises((ValueError, TypeError)): + device.update_ExcitationFrequency(value) + device.write_command_string.assert_not_called() + + +def test_mixed_legacy_and_block_configuration_rejected_before_writes(): + device = EIT_16_32_64_128(16) + device.cMessageParser = Mock() + device.write_command_string = Mock() + setup = EitMeasurementSetup( + 1, 16, [EitFrequencyBlock(1000, 2000, 2)], 2, 0.001, 0, 1, 1, n_freq=2 + ) + with pytest.raises(ValueError, match="legacy"): + device.SetMeasurementSetup(setup) + device.write_command_string.assert_not_called() diff --git a/tests/test_usb_message_parser.py b/tests/test_usb_message_parser.py index afb9578..ab1ff1f 100644 --- a/tests/test_usb_message_parser.py +++ b/tests/test_usb_message_parser.py @@ -1,11 +1,18 @@ from unittest.mock import Mock +import struct import numpy as np +import pytest -from sciopy.sciopy_dataclasses import EITFrame +from sciopy.sciopy_dataclasses import ( + EITFrame, + EitFrequencyBlock, + EitMeasurementSetup, +) from sciopy.usb_message_parser import ( MessageParser, describe_message, + get_data_as_matrix, load_eit_frames, save_data_frame, ) @@ -92,6 +99,272 @@ def test_acknowledgement_is_labelled_with_its_pending_command(capsys): assert parser.pending_command is None +def test_parser_allocates_buffer_for_all_frequency_settings(): + setup = EitMeasurementSetup( + burst_count=2, + n_el=16, + exc_freq=[ + EitFrequencyBlock( + f_min=1_000, + f_max=10_000, + f_count=10, + f_type="lin", + ), + EitFrequencyBlock( + f_min=10_000, + f_max=100_000, + f_count=20, + f_type="log", + ), + ], + framerate=2.0, + amplitude=0.001, + inj_skip=0, + gain=1, + adc_range=1, + ) + + parser = MessageParser( + Mock(), + eitsetup=setup, + ) + + assert parser.iNumFreqSettings == 30 + assert parser.iNumExcitationSettings == 16 + assert parser.iLenDataperFrame == 16 * 30 * 16 + assert len(parser.CurrentFrame.ppcData) == 16 * 30 * 16 + + assert np.array_equal( + parser.CurrentFrame.frequency_stgs, + np.concatenate( + [ + np.linspace(1000, 10000, 10), + np.geomspace(10000, 100000, 20), + ] + ), + ) + + +def test_get_data_as_matrix_preserves_single_frequency_shape(): + frame = EITFrame( + n_el=3, + excitation_stgs=np.array( + [ + [1, 2], + [2, 3], + ] + ), + frequency_stgs=np.array([1]), + timestamp1=0, + timestamp2=0, + timestamp_pc=0, + ppcData=np.arange( + 6, + dtype=np.complex128, + ), + ) + + result = get_data_as_matrix([frame]) + + assert result.shape == ( + 1, + 2, + 3, + ) + + assert np.array_equal( + result[0], + np.array( + [ + [0, 1, 2], + [3, 4, 5], + ] + ), + ) + + +def test_get_data_as_matrix_adds_frequency_axis_for_sweep(): + frame = EITFrame( + n_el=3, + excitation_stgs=np.array( + [ + [1, 2], + [2, 3], + ] + ), + frequency_stgs=np.array([1, 2]), + timestamp1=0, + timestamp2=0, + timestamp_pc=0, + ppcData=np.arange( + 12, + dtype=np.complex128, + ), + ) + + result = get_data_as_matrix([frame]) + + assert result.shape == ( + 1, + 2, + 2, + 3, + ) + + assert np.array_equal( + result[0, 0, 0], + [0, 1, 2], + ) + + assert np.array_equal( + result[0, 0, 1], + [3, 4, 5], + ) + + assert np.array_equal( + result[0, 1, 0], + [6, 7, 8], + ) + + assert np.array_equal( + result[0, 1, 1], + [9, 10, 11], + ) + + +def test_parser_collects_all_frequency_rows_into_one_frame(): + setup = EitMeasurementSetup( + burst_count=1, + n_el=16, + exc_freq=[ + EitFrequencyBlock( + f_min=1_000, + f_max=1_000, + f_count=1, + f_type="lin", + ), + EitFrequencyBlock( + f_min=2_000, + f_max=2_000, + f_count=1, + f_type="lin", + ), + ], + framerate=2.0, + amplitude=0.001, + inj_skip=0, + gain=1, + adc_range=1, + ) + + parser = MessageParser( + Mock(), + eitsetup=setup, + ) + + def make_data_message( + excitation_out, + excitation_in, + frequency_row, + ): + message = [ + 0xB4, + 0x84, + 0x01, + excitation_out, + excitation_in, + ] + + message.extend( + frequency_row.to_bytes( + 2, + byteorder="big", + ) + ) + + message.extend( + struct.pack( + ">f", + 0.0, + ) + ) + + for channel in range(16): + value = excitation_out * 100 + frequency_row * 10 + channel + + message.extend( + struct.pack( + ">f", + float(value), + ) + ) + + message.extend( + struct.pack( + ">f", + float(-value), + ) + ) + + message.append(0xB4) + + return message + + message_count = 0 + + for excitation_out in range( + 1, + 17, + ): + excitation_in = (excitation_out % 16) + 1 + + for frequency_row in ( + 1, + 2, + ): + parser.interpret_data_input( + make_data_message( + excitation_out, + excitation_in, + frequency_row, + ) + ) + + message_count += 1 + + if message_count < 32: + assert len(parser.ppcData) == 0 + + assert len(parser.ppcData) == 1 + + frame = parser.ppcData[0] + + assert frame.excitation_stgs.shape == ( + 16, + 2, + ) + + assert np.array_equal( + frame.frequency_stgs, + [1000, 2000], + ) + + result = get_data_as_matrix(parser.ppcData) + + assert result.shape == ( + 1, + 16, + 2, + 16, + ) + + assert result[0, 0, 0, 0] == (110 - 110j) + + assert result[0, 0, 1, 0] == (120 - 120j) + + assert result[0, 15, 1, 15] == (1_635 - 1_635j) + + def test_saved_eit_frame_round_trips_and_ignores_other_files(tmp_path): frame = EITFrame( n_el=16, @@ -110,3 +383,68 @@ def test_saved_eit_frame_round_trips_and_ignores_other_files(tmp_path): assert len(loaded) == 1 assert loaded[0].n_el == 16 assert np.array_equal(loaded[0].ppcData, frame.ppcData) + + +@pytest.mark.parametrize("n_el", [16, 32]) +def test_multiple_skips_and_frequency_blocks_parse_in_configured_order(n_el): + setup = EitMeasurementSetup( + 1, n_el, [EitFrequencyBlock(1000, 2000, 2)], 2, 0.001, [0, 2], 1, 1 + ) + parser = MessageParser(Mock(), setup) + expected = np.arange(n_el * 2 * 2 * n_el).reshape(n_el * 2, 2, n_el) + index = 0 + for skip in (0, 2): + for out in range(1, n_el + 1): + for freq in (1, 2): + for group in range(1, n_el // 16 + 1): + msg = [0xB4, 0x84, group, out, ((out + skip) % n_el) + 1] + msg.extend(struct.pack(">Hf", freq, 0)) + for _ in range(16): + msg.extend(struct.pack(">ff", index, -index)) + index += 1 + msg.append(0xB4) + parser.interpret_data_input(msg) + result = get_data_as_matrix(parser.ppcData) + np.testing.assert_array_equal(result[0], expected - 1j * expected) + assert result.shape == (1, n_el * 2, 2, n_el) + np.testing.assert_array_equal(parser.ppcData[0].frequency_stgs, [1000, 2000]) + assert parser._completed_frames == 1 + + +def test_measurement_read_tolerates_empty_reads_and_stops_after_saved_frame( + monkeypatch, tmp_path +): + setup = EitMeasurementSetup(1, 16, 1000, 2, 0.001, 0, 1, 1) + parser = MessageParser(Mock(), setup) + messages = [] + for out in range(1, 17): + msg = [0xB4, 0x84, 1, out, out % 16 + 1] + msg.extend(struct.pack(">Hf", 1, 0)) + msg.extend(struct.pack(">ff", 1, -1) * 16) + msg.append(0xB4) + messages.append(msg) + parser.device_read = Mock(side_effect=[[], [1]]) + parser.parse_received_bytes = Mock(return_value=messages) + monkeypatch.setattr("sciopy.usb_message_parser.time.sleep", lambda _: None) + result = parser.read_usb_till_timeout(True, True, str(tmp_path) + "/") + assert result == [] + assert parser.device_read.call_count == 2 + assert parser._completed_frames == 1 + assert len(list(tmp_path.glob("*.npz"))) == 1 + + +def test_measurement_read_expires_when_idle(monkeypatch): + parser = MessageParser(Mock(), EitMeasurementSetup(1, 16, 1000, 2, 0.001, 0, 1, 1)) + parser.device_read = Mock(return_value=[]) + times = iter([0, 1, 6]) + monkeypatch.setattr("sciopy.usb_message_parser.time.monotonic", lambda: next(times)) + monkeypatch.setattr("sciopy.usb_message_parser.time.sleep", lambda _: None) + assert parser.read_usb_till_timeout() == [] + assert parser.device_read.call_count == 2 + + +def test_command_response_keeps_immediate_timeout_with_setup(): + parser = MessageParser(Mock(), EitMeasurementSetup(1, 16, 1000, 2, 0.001, 0, 1, 1)) + parser.device_read = Mock(return_value=[]) + assert parser.read_usb_till_timeout(bStartReset=False) == [] + parser.device_read.assert_called_once()