Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion examples/EIT-16-256-Ch.ipynb
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,7 @@
"setup = EitMeasurementSetup(\n",
" burst_count=1,\n",
" n_el=n_el,\n",
" # exc_freq={'f_min': 125_000, 'f_max':1_000_000, 'steps':10, 'sweep': \"LOG\"},\n",
" exc_freq=125_000,\n",
" framerate=3,\n",
" amplitude=0.01,\n",
Expand Down Expand Up @@ -380,7 +381,7 @@
],
"metadata": {
"kernelspec": {
"display_name": "Python 3 (ipykernel)",
"display_name": "sciopy",
"language": "python",
"name": "python3"
},
Expand Down
16 changes: 7 additions & 9 deletions examples/ISX-3.ipynb
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
"import cmath\n",
"\n",
"import matplotlib.pyplot as plt\n",
"import numpy as np\n",
"\n",
"from sciopy import ISX_3, EisMeasurementSetup, available_serial_ports"
]
Expand Down Expand Up @@ -373,16 +374,13 @@
},
"outputs": [],
"source": [
"mux_data = {}\n",
"\n",
"for name, (counter, reference, working_sense, working) in mux_routings.items():\n",
" isx.SetExtensionPortChannel(counter, reference, working_sense, working)\n",
" isx.GetExtensionPortChannel() # Read back the active C/R/WS/W routing.\n",
" points = isx.StartStopMeasurement()\n",
" if not points:\n",
" raise RuntimeError(f\"No measurements received for {name}.\")\n",
" mux_data[name] = points\n",
" print(f\"{name}: received {len(points)} points\")"
"\n",
"points = isx.StartStopMeasurement()\n",
"\n",
"mux_data = {f\"routing {i}\": c\n",
" for i, c in enumerate(np.array_split(points, len( mux_routings.keys())), start=1)}"
]
},
{
Expand Down Expand Up @@ -438,7 +436,7 @@
],
"metadata": {
"kernelspec": {
"display_name": "Python 3 (ipykernel)",
"display_name": "sciopy",
"language": "python",
"name": "python3"
},
Expand Down
153 changes: 128 additions & 25 deletions sciopy/EIT_16_32_64_128.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
from .com_util import (
clTbt_dp,
clTbt_sp,
clTbt_u16,
)

import numpy as np
Expand Down Expand Up @@ -264,32 +265,109 @@ def update_FrameRate(self, framerate):
)
self.print_msg = False

def update_ExcitationFrequency(self, exc_freq):
def update_ExcitationFrequencies(
self,
f_min: float,
f_max: float = None,
f_count: int = 1,
f_scale: str = "lin",
):
"""
update_ExcitationFrequencies _summary_
Configures a single excitation frequency or a full frequency sweep.

Sends the "Excitation Frequencies" option (0x04) of the "Set
Measurement Setup" command (0xB0), as specified by the Sciospec
System Message Protocol for the EIT-16/32/64/128 device family:

[CT] 0x0C 0x04 [f_min] [f_max] [f_count] [f_scale] [CT]

f_min, f_max : IEEE-754 single precision float, big-endian
(4 bytes each). Minimum / maximum excitation
frequency in Hz (100 Hz to 1 MHz).
f_count : unsigned 16-bit integer, big-endian (2 bytes).
Number of frequency points measured between
f_min and f_max (inclusive). f_count=1 selects a
single-frequency measurement at f_min.
f_scale : 1 byte. 0x00 = linear, 0x01 = logarithmic
distribution of the frequency points between
f_min and f_max.

Parameters
----------
exc_freq int
frequency to be set from 100 Hz to 1 MHz
"""
f_min : float
Start (or only) excitation frequency in Hz.
f_max : float, optional
Stop excitation frequency in Hz. Defaults to `f_min`, i.e. a
single-frequency measurement.
f_count : int, optional
Number of frequency points in the sweep. Defaults to 1.
f_scale : str, optional
``"lin"`` for a linear sweep or ``"log"`` for a logarithmic
sweep. Defaults to ``"lin"``. Ignored when `f_count` is 1.

Side Effects
------------
- Updates `self.setup` (exc_freq, exc_freq_max, n_freq, freq_scale)
and re-syncs the message parser's frame layout, when a
measurement setup is present, so subsequently received frames are
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.")

if self.setup is not None:
self.setup.exc_freq = f_min
self.setup.exc_freq_max = f_max
self.setup.n_freq = f_count
self.setup.freq_scale = f_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] [ftype] [CT]
# [CT] 0C 04 [fmin] [fmax] [fcount] [fscale] [CT]
self.print_msg = True
f_min = clTbt_sp(exc_freq)
f_max = clTbt_sp(exc_freq)
f_count = [0, 1]
f_type = [0] # linear/log
# bytearray
self.write_command_string(
bytearray(
list(
np.concatenate([[176, 12, 4], f_min, f_max, f_count, f_type, [176]])
np.concatenate(
[
[176, 12, 4],
clTbt_sp(f_min),
clTbt_sp(f_max),
clTbt_u16(f_count),
[scale_options[f_scale]],
[176],
]
)
)
)
)
self.print_msg = False

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)``.

Parameters
----------
exc_freq : int or float
frequency to be set from 100 Hz to 1 MHz
"""
self.update_ExcitationFrequencies(f_min=exc_freq, f_count=1)

def SetMeasurementSetup(self, setup: EitMeasurementSetup):
"""
Configures the ScioSpec device measurement setup according to the provided EitMeasurementSetup dataclass.
Expand Down Expand Up @@ -385,19 +463,15 @@ def SetMeasurementSetup(self, setup: EitMeasurementSetup):
list(np.concatenate([[176, 5, 3], clTbt_sp(setup.framerate), [176]]))
)
)
# Set frequencies:
# [CT] 0C 04 [fmin] [fmax] [fcount] [ftype] [CT]
f_min = clTbt_sp(setup.exc_freq)
f_max = clTbt_sp(setup.exc_freq)
f_count = [0, 1]
f_type = [0] # linear/log
# bytearray
self.write_command_string(
bytearray(
list(
np.concatenate([[176, 12, 4], f_min, f_max, f_count, f_type, [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,
)

# Set injection config
Expand Down Expand Up @@ -503,6 +577,10 @@ def StartStopMeasurement(
bDeleteData: bool = False,
sSavePath: str = "C/",
bResultsFolder=False,
f_min: float = None,
f_max: float = None,
f_count: int = None,
f_scale: str = "lin",
):
"""
Starts and stops a measurement process using the configured serial protocol (HS or FS).
Expand All @@ -524,9 +602,24 @@ def StartStopMeasurement(
bSaveData=True, measured data is saved and then removed from RAM
sSavePath (str): Specifies the sPath where the measured data is saved.
bResultsFolder (bool): Specifies if additionally a folder in sSavePath is created to store the data in
f_min (float, optional): When given, (re)configures the device's excitation
frequency/sweep via `update_ExcitationFrequencies` before starting the
measurement. Omit to measure with whatever sweep is already configured
(e.g. via `SetMeasurementSetup`).
f_max (float, optional): Stop frequency of the sweep in Hz. Defaults to
`f_min` (single-frequency measurement) when `f_min` is given but
`f_max` is not.
f_count (int, optional): Number of frequency points in the sweep. Defaults
to `self.setup.n_freq` (or 1) when `f_min` is given but `f_count` is not.
f_scale (str, optional): "lin" or "log" distribution of the sweep's
frequency points. Defaults to "lin".

Returns:
list or matrix: The measurement data in the format specified by `return_as`.
When a frequency sweep (`f_count` / `setup.n_freq` > 1) is active and
`return_as="pot_mat"`, the returned matrix has shape
(n_frames, n_excitations, n_freq, n_el); otherwise it keeps the previous
(n_frames, n_excitations, n_el) shape.
"""

if self.cMessageParser is None:
Expand All @@ -536,6 +629,16 @@ def StartStopMeasurement(
if return_as not in {"hex", "pot_mat", "eitframe"}:
raise ValueError("return_as must be 'hex', 'pot_mat' or 'eitframe'.")

if f_min is not None:
# (Re-)configure a single frequency or a full sweep right before
# the measurement starts.
self.update_ExcitationFrequencies(
f_min=f_min,
f_max=f_max,
f_count=f_count if f_count is not None else (self.setup.n_freq or 1),
f_scale=f_scale,
)

# Start measurement
self.cMessageParser.clear_out_data()
sCurrentPath = make_results_folder(
Expand Down
4 changes: 3 additions & 1 deletion sciopy/ISX_3.py
Original file line number Diff line number Diff line change
Expand Up @@ -568,7 +568,9 @@ def GetFE_Settings(self):
return self.send_command(0xB1)

def SetExtensionPortChannel(self, counter, reference, working_sense, working):
"""Configure C, R, WS, and W selections on the extension port."""
"""Configure C, R, WS, and W selections on the extension port.
It adds to the current config instead of creating a new extension port channel.
"""
return self.send_command(0xB2, [counter, reference, working_sense, working])

def GetExtensionPortChannel(self):
Expand Down
13 changes: 13 additions & 0 deletions sciopy/com_util.py
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,19 @@ def clTbt_dp(val: float) -> list:
return [int(ele) for ele in struct.pack(">d", val)]


def clTbt_u16(val: int) -> list:
"""
clTbt_u16 converts an unsigned integer to a list of 2 big-endian bytes.

Used e.g. for the frequency-point count of the "Excitation Frequencies"
measurement setup option (0xB0 0x0C 0x04), which the device expects as
an unsigned 16-bit big-endian integer.
"""
if not 0 <= val <= 0xFFFF:
raise ValueError(f"val must fit in an unsigned 16-bit integer, got {val}.")
return [int(bt) for bt in struct.pack(">H", val)]


def reshape_full_message_in_bursts(lst: list, ssms: EitMeasurementSetup) -> np.ndarray:
"""
Takes the full message buffer and splits this message depeding on the measurement configuration into the
Expand Down
25 changes: 21 additions & 4 deletions sciopy/sciopy_dataclasses.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,12 +13,25 @@ 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.
exc_freq (int or float): Excitation frequency in Hz. 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.
gain (int): Amplifier gain setting.
adc_range (int): Analog-to-digital converter range setting.
mea_mode (str): Measurement mode, see `update_measurement_mode`.
mea_mode_boundary (str): Channel-group boundary behavior, see
`update_measurement_mode`.
exc_freq_max (int, float or None): Stop (maximum) frequency of a
frequency sweep in Hz. Ignored when ``n_freq`` is 1. Defaults to
``exc_freq`` (single-frequency measurement) when left as ``None``.
n_freq (int): Number of frequency points measured between
``exc_freq`` and ``exc_freq_max`` (inclusive). ``1`` selects a
single-frequency measurement at ``exc_freq``.
freq_scale (str): Distribution of frequency points across the sweep,
either ``"lin"`` (linear) or ``"log"`` (logarithmic). Ignored
when ``n_freq`` is 1.
"""

burst_count: int
Expand All @@ -31,7 +44,9 @@ class EitMeasurementSetup:
adc_range: int
mea_mode: str = "singleended"
mea_mode_boundary: str = "internal"
# TBD: lin/log/sweep
exc_freq_max: Union[int, float, None] = None
n_freq: int = 1
freq_scale: str = "lin"


@dataclass
Expand Down Expand Up @@ -225,7 +240,9 @@ class EITFrame:
----------
n_el = Number of used electrodes
excitation_stgs : np.array [[int1, int2]] , Features the [ESout, ESin] injection electrodes
frequency_stgs : List[str] # todo
frequency_stgs : np.array of the excitation frequencies used in this frame [Hz],
e.g. [f] for a single-frequency measurement or
[f_min, ..., f_max] (length n_freq) for a frequency sweep.
timestamp1 : int Timestamp of the very first measured channel group in this frame, milli seconds?
timestamp2 : int Timestamp of the very last measured channel group in this frame, milli seconds?
timestamp_pc : int Timestamp of the receiving computer for further data synchronisation from datetime.now().
Expand All @@ -238,7 +255,7 @@ class EITFrame:

n_el: int # Number of used electrodes
excitation_stgs: npt.NDArray[int] # Num Excitation Settings X 2
frequency_stgs: npt.NDArray[int] # List of Frequency-Sweep Settings,
frequency_stgs: npt.NDArray[float] # Excitation frequencies of the sweep [Hz]
timestamp1: int # [ms]
timestamp2: int
timestamp_pc: int
Expand Down
Loading
Loading