Skip to content
Open
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
7 changes: 4 additions & 3 deletions examples/example_device_config.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,9 @@ TxConfiguration:
channel_max_amplitude: [200, 6000, 6000, 6000]
# Filter configuration of each transmit channel
channel_filter_type: [0, 2, 2, 2]
# Configure which channels are terminated into 50 ohm impedance (true == 50 ohm), list length must at least match the number of enabled channels
channel_terminated_50ohm: [True, False, False, False]
# Configure if RF/gradients are terminated into 50 ohm impedance (true == 50 ohm)
rf_terminated_50ohm: True
gradients_terminated_50ohm: False
# Calculate grad_to_volt per gradient channel:
# Gradient efficiency in T/m/A
gradient_efficiency: [0.37e-3, 0.451e-3, 0.4e-3]
Expand All @@ -33,7 +34,7 @@ RxConfiguration:
# Could be extended to a list
device_path: "/dev/spcm0"
# Number of channels
max_available_channels: 4
max_available_channels: 8
# Device sampling rate in MHz
sampling_rate: 20
# Enable the receive channels
Expand Down
100 changes: 39 additions & 61 deletions src/console/interfaces/device_configuration.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,15 @@
"""Implementation of the device configuration models."""
from pydantic import BaseModel, Field, field_validator, model_validator
from typing import Annotated

# Specify the number of gradient channels
NUM_GRADIENTS = 3
from pydantic import BaseModel, Field, model_validator

# Ensure that max. amplitude is within 1 and 6000 mV.
# Limits may depend on specific configuration of spectrum cards.
AmplitudeInt = Annotated[int, Field(ge=1, le=6000)]
# Ensure valid filter type, channel filter type must be 0, 1, 2 or 3.
# See spectrum instrumentation manual for reference.
# This may depend on the specific card configuration.
FilterTypeInt = Annotated[int, Field(ge=0, le=3)]

class TxConfiguration(BaseModel):
"""Transmit device configuration."""
Expand All @@ -12,50 +18,14 @@ class TxConfiguration(BaseModel):

device_path: str = Field(..., strict=True)
sampling_rate: int = Field(..., strict=True)
max_available_channels: int = Field(..., strict=True)
channel_max_amplitude: list[int] = Field(..., strict=True)
channel_filter_type: list[int] = Field(..., strict=True)
channel_terminated_50ohm: list[bool] = Field(..., strict=True)
gradient_efficiency: list[float] = Field(..., min_length=NUM_GRADIENTS, max_length=NUM_GRADIENTS, strict=True)
gpa_gain: list[float] = Field(..., min_length=NUM_GRADIENTS, max_length=NUM_GRADIENTS, strict=True)
channel_max_amplitude: tuple[AmplitudeInt, AmplitudeInt, AmplitudeInt, AmplitudeInt] = Field(...)
channel_filter_type: tuple[FilterTypeInt, FilterTypeInt, FilterTypeInt, FilterTypeInt] = Field(...)
rf_terminated_50ohm: bool = Field(..., strict=True)
gradients_terminated_50ohm: bool = Field(..., strict=True)
gradient_efficiency: tuple[float, float, float] = Field(...)
gpa_gain: tuple[float, float, float] = Field(...)
rf_to_mvolt: float = Field(..., strict=True)

@model_validator(mode="after")
def validate_lists(self) -> "TxConfiguration":
"""Validate the length of device-specific lists."""
list_fields: dict[str, list[int] | list[bool]] = {
"channel_max_amplitude": self.channel_max_amplitude,
"channel_filter_type": self.channel_filter_type,
"channel_terminated_50ohm": self.channel_terminated_50ohm,
}
errors = []
for name, val in list_fields.items():
if len(val) < self.max_available_channels:
_err = f"Field '{name}': \
length {len(val)} is less than max_available_channels {self.max_available_channels}"
errors.append(_err)
if errors:
raise ValueError("; ".join(errors))
return self

@field_validator("channel_max_amplitude")
@classmethod
def validate_max_amplitude(cls, values: list[int]) -> list[int]:
"""Ensure that max. amplitude is within 1 and 6000 mV."""
if not all(val in range(1, 6001) for val in values):
msg = "Channel max amplitude must be between 1 and 6000 mV."
raise ValueError(msg)
return values

@field_validator("channel_filter_type")
@classmethod
def validate_filter_type(cls, values: list[int]) -> list[int]:
"""Ensure that max. amplitude is within 1 and 6000 mV."""
if not all(val in range(4) for val in values):
msg = "Channel filter type must be 0, 1, 2 or 3."
raise ValueError(msg)
return values


class RxConfiguration(BaseModel):
"""Receive device configuration."""
Expand All @@ -65,26 +35,34 @@ class RxConfiguration(BaseModel):
device_path: str = Field(..., strict=True)
sampling_rate: int = Field(..., strict=True)
max_available_channels: int = Field(..., strict=True)
channel_enable: list[bool] = Field(..., strict=True)
channel_max_amplitude: list[int] = Field(..., strict=True)
channel_terminated_50ohm: list[bool] = Field(..., strict=True)
channel_enable: tuple[bool, ...] = Field(...)
channel_max_amplitude: tuple[AmplitudeInt, ...] = Field(...)
channel_terminated_50ohm: tuple[bool, ...] = Field(...)

@model_validator(mode="after")
def validate_lists(self) -> "RxConfiguration":
"""Validate the length of device-specific lists."""
list_fields: dict[str, list[int] | list[bool]] = {
"channel_max_amplitude": self.channel_max_amplitude,
"channel_enable": self.channel_enable,
"channel_terminated_50ohm": self.channel_terminated_50ohm,
}
def validate_channel_lengths(self) -> "RxConfiguration":
"""Validate channel lengths.

This ensures that `max_available_channels` is length of `channel_enable`,
`channel_max_amplitude` and `channel_terminated_50ohm`.
"""
target_len = self.max_available_channels
dependent_fields = [
"channel_enable",
"channel_max_amplitude",
"channel_terminated_50ohm",
]
errors = []
for name, val in list_fields.items():
if len(val) < self.max_available_channels:
_err = f"Field '{name}'\
length {len(val)} is less than max_available_channels {self.max_available_channels}"
errors.append(_err)
for field_name in dependent_fields:
current_val = getattr(self, field_name)
if len(current_val) != target_len:
errors.append(
f"{field_name} length ({len(current_val)}) "
f"must be exactly {target_len}"
)
if errors:
raise ValueError("; ".join(errors))
# Raising ValueError here is the standard Pydantic way
raise ValueError(" ; ".join(errors))
return self


Expand Down
12 changes: 10 additions & 2 deletions src/console/interfaces/unrolled_sequence.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,22 +28,30 @@ class UnrolledSequence:
rx_data: list[RxData]
"""List containing the data and metadata of all receive events"""

gpa_gain: list[float]
gpa_gain: tuple[float, float, float]
"""The gradient waveforms in pulseq are defined in Hz/m.
The translation to mV is calculated by 1e3 / (gyro * gpa_gain * grad_efficiency).
The gpa gain is given in V/A and accounts for the voltage required to generate an output of 1A.
The gyromagnetic ratio defined by 42.58e6 MHz/T."""

gradient_efficiency: list[float]
gradient_efficiency: tuple[float, float, float]
"""The gradient waveforms in pulseq are defined in Hz/m.
The translation to mV is calculated by 1e3 / (gyro * gpa_gain * grad_efficiency).
The gradient efficiency is given in mT/m/A and accounts for the gradient field which is generated per 1A.
The gyromagnetic ratio defined by 42.58e6 MHz/T."""

gradient_output_limits: tuple[int, int, int]
"""Integer limits for each gradient output channel in mV. Gradient output limits already contain scaling
from channel termination into high impedance or 50 ohms respectively."""

rf_to_mvolt: float
"""If sequence values are given as float values, they can be interpreted as output voltage [mV] directly.
This conversion factor represents the scaling from original pulseq RF values [Hz] to card output voltage."""

rf_output_limit: int
"""Integer limit of rf output channel in mV. RF output limits already contains scaling
from channel termination into high impedance or 50 ohms respectively."""

dwell_time: float
"""Dwell time of the spectrum card replay data (unrolled sequence).
Defines the distance in time between to sample points.
Expand Down
Loading