nqrduck/quackseq
1
0
Fork 0
mirror of https://github.com/nqrduck/quackseq.git synced 2024-09-16 14:00:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Initial structure.

Browse source
This commit is contained in:
jupfi 2024-05-28 20:35:56 +02:00
commit e13ea949bc
17 changed files with 1986 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
.gitignore vendored Normal file
View file

@ -0,0 +1,24 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.pyc
*$py.class
# Distribution / packaging
dist/
build/
*.egg-info/
# IDE-specific files
.idea/
.vscode/
# Logs
*.log
# Virtual environments
venv/
# Other
*.DS_Store
*.pos
*.quack

5
CHANGELOG.md Normal file
View file

@ -0,0 +1,5 @@
# Changelog
## Version 0.0.1 (15-04-2024)
- Initial release

20
LICENSE Normal file
View file

@ -0,0 +1,20 @@
MIT License
Copyright (c) 2023-2024 jupfi
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

1
README.md Normal file
View file

@ -0,0 +1 @@
# quackseq

59
pyproject.toml Normal file
View file

@ -0,0 +1,59 @@
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.metadata]
allow-direct-references = true
[project]
name = "quackseq"
version = "0.0.1"
authors = [
{ name="jupfi", email="support@nqrduck.cool" },
]
description = "Simple Python script to perform magnetic resonance spectroscopy experiments."
readme = "README.md"
license = { file="LICENSE" }
requires-python = ">=3.10"
classifiers = [
"Programming Language :: Python :: 3",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
]
dependencies = [
"numpy",
"scipy",
]
[project.optional-dependencies]
dev = [
"black",
"pydocstyle",
"pyupgrade",
"ruff",
]
[tool.ruff]
[tool.ruff.lint]
extend-select = [
"UP", # pyupgrade
"D", # pydocstyle
]
[tool.ruff.lint.per-file-ignores]
"__init__.py" = ["F401"]
[tool.ruff.lint.pydocstyle]
convention = "google"
[project.urls]
"Homepage" = "https://nqrduck.cool"
"Bug Tracker" = "https://github.com/nqrduck/quackseq/issues"
"Source Code" = "https://github.com/nqrduck/quackseq"
[tool.hatch.build.targets.wheel]
packages = ["src/quackseq"]

0
src/quackseq/__init__.py Normal file
View file

95
src/quackseq/event.py Normal file
View file

@ -0,0 +1,95 @@
import logging
from collections import OrderedDict
from quackseq.pulseparameters import Option
from quackseq.helpers import UnitConverter
logger = logging.getLogger(__name__)
class Event:
"""An event is a part of a pulse sequence. It has a name and a duration and different parameters that have to be set.
Args:
name (str): The name of the event
duration (str): The duration of the event
Attributes:
name (str): The name of the event
duration (str): The duration of the event
parameters (OrderedDict): The parameters of the event
"""
def __init__(self, name: str, duration: str) -> None:
"""Initializes the event."""
self.parameters = OrderedDict()
self.name = name
self.duration = duration
def add_parameter(self, parameter) -> None:
"""Adds a parameter to the event.
Args:
parameter: The parameter to add
"""
self.parameters.append(parameter)
def on_duration_changed(self, duration: str) -> None:
"""This method is called when the duration of the event is changed.
Args:
duration (str): The new duration of the event
"""
logger.debug("Duration of event %s changed to %s", self.name, duration)
self.duration = duration
@classmethod
def load_event(cls, event, pulse_parameter_options):
"""Loads an event from a dict.
The pulse paramter options are needed to load the parameters
and determine if the correct spectrometer is active.
Args:
event (dict): The dict with the event data
pulse_parameter_options (dict): The dict with the pulse parameter options
Returns:
Event: The loaded event
"""
obj = cls(event["name"], event["duration"])
for parameter in event["parameters"]:
for pulse_parameter_option in pulse_parameter_options.keys():
# This checks if the pulse paramter options are the same as the ones in the pulse sequence
if pulse_parameter_option == parameter["name"]:
pulse_parameter_class = pulse_parameter_options[
pulse_parameter_option
]
obj.parameters[pulse_parameter_option] = pulse_parameter_class(
parameter["name"]
)
# Delete the default instances of the pulse parameter options
obj.parameters[pulse_parameter_option].options = []
for option in parameter["value"]:
obj.parameters[pulse_parameter_option].options.append(
Option.from_json(option)
)
return obj
@property
def duration(self):
"""The duration of the event."""
return self._duration
@duration.setter
def duration(self, duration: str):
# Duration needs to be a positive number
try:
duration = UnitConverter.to_float(duration)
except ValueError:
raise ValueError("Duration needs to be a number")
if duration < 0:
raise ValueError("Duration needs to be a positive number")
self._duration = duration

349
src/quackseq/functions.py Normal file
View file

@ -0,0 +1,349 @@
"""A module that contains functions that can be used in various modules. This used to be part of the nqrduck-spectrometer module."""
from __future__ import annotations
import logging
import numpy as np
import sympy
logger = logging.getLogger(__name__)
class Function:
"""A function that can be used as a pulse parameter.
This class is the base class for all functions that can be used as pulse parameters. Functions can be used for pulse shapes, for example.
Args:
expr (str | sympy.Expr): The expression of the function.
Attributes:
name (str): The name of the function.
parameters (list): The parameters of the function.
expr (sympy.Expr): The sympy expression of the function.
resolution (float): The resolution of the function in seconds.
start_x (float): The x value where the evalution of the function starts.
end_x (float): The x value where the evalution of the function ends.
"""
name: str
parameters: list
expression: str | sympy.Expr
resolution: float
start_x: float
end_x: float
subclasses = []
def __init_subclass__(cls, **kwargs):
"""Registers the subclass."""
super().__init_subclass__(**kwargs)
cls.subclasses.append(cls)
def __init__(self, expr) -> None:
"""Initializes the function."""
self.parameters = []
self.expr = expr
self.resolution = 1 / 30.72e6
self.start_x = -1
self.end_x = 1
def get_time_points(self, pulse_length: float) -> np.ndarray:
"""Returns the time domain points for the function with the given pulse length.
Args:
pulse_length (float): The pulse length in seconds.
Returns:
np.ndarray: The time domain points.
"""
# Get the time domain points
n = int(pulse_length / self.resolution)
t = np.linspace(0, float(pulse_length), n)
return t
def evaluate(self, pulse_length: float, resolution: float = None) -> np.ndarray:
"""Evaluates the function for the given pulse length.
Args:
pulse_length (float): The pulse length in seconds.
resolution (float, optional): The resolution of the function in seconds. Defaults to None.
Returns:
np.ndarray: The evaluated function.
"""
if resolution is None:
resolution = self.resolution
n = int(pulse_length / resolution)
t = np.linspace(self.start_x, self.end_x, n)
x = sympy.symbols("x")
found_variables = dict()
# Create a dictionary of the parameters and their values
for parameter in self.parameters:
found_variables[parameter.symbol] = parameter.value
final_expr = self.expr.subs(found_variables)
# If the expression is a number (does not depend on x), return an array of that number
if final_expr.is_number:
return np.full(t.shape, float(final_expr))
f = sympy.lambdify([x], final_expr, "numpy")
return f(t)
def get_tdx(self, pulse_length: float) -> np.ndarray:
"""Returns the time domain points and the evaluated function for the given pulse length.
Args:
pulse_length (float): The pulse length in seconds.
Returns:
np.ndarray: The time domain points.
"""
n = int(pulse_length / self.resolution)
t = np.linspace(self.start_x, self.end_x, n)
return t
def get_pulse_amplitude(
self, pulse_length: float, resolution: float = None
) -> np.array:
"""Returns the pulse amplitude in the time domain.
Args:
pulse_length (Float): The pulse length in seconds.
resolution (float, optional): The resolution of the function in seconds. Defaults to None.
Returns:
np.array: The pulse amplitude.
"""
return self.evaluate(pulse_length, resolution=resolution)
def add_parameter(self, parameter: Function.Parameter) -> None:
"""Adds a parameter to the function.
Args:
parameter (Function.Parameter): The parameter to add.
"""
self.parameters.append(parameter)
def to_json(self) -> dict:
"""Returns a json representation of the function.
Returns:
dict: The json representation of the function.
"""
return {
"name": self.name,
"class" : self.__class__.__name__,
"parameters": [parameter.to_json() for parameter in self.parameters],
"expression": str(self.expr),
"resolution": self.resolution,
"start_x": self.start_x,
"end_x": self.end_x,
}
@classmethod
def from_json(cls, data: dict) -> Function:
"""Creates a function from a json representation.
Args:
data (dict): The json representation of the function.
Returns:
Function: The function.
"""
logger.debug(f"Data: {data}")
for subclass in cls.subclasses:
logger.debug("Checking subclass %s", subclass)
logger.debug("Subclass name %s", subclass.__name__)
if subclass.__name__ == data["class"]:
cls = subclass
logger.debug("Found subclass %s", cls)
break
obj = cls()
obj.expr = data["expression"]
obj.name = data["name"]
obj.resolution = data["resolution"]
obj.start_x = data["start_x"]
obj.end_x = data["end_x"]
obj.parameters = []
for parameter in data["parameters"]:
obj.add_parameter(Function.Parameter.from_json(parameter))
return obj
@property
def expr(self):
"""The sympy expression of the function."""
return self._expr
@expr.setter
def expr(self, expr):
if isinstance(expr, str):
try:
self._expr = sympy.sympify(expr)
except ValueError:
logger.error(f"Could not convert {expr} to a sympy expression")
raise SyntaxError(f"Could not convert {expr} to a sympy expression")
elif isinstance(expr, sympy.Expr):
self._expr = expr
@property
def resolution(self):
"""The resolution of the function in seconds."""
return self._resolution
@resolution.setter
def resolution(self, resolution):
try:
self._resolution = float(resolution)
except ValueError:
logger.error("Could not convert %s to a float", resolution)
raise SyntaxError(f"Could not convert {resolution} to a float")
@property
def start_x(self):
"""The x value where the evalution of the function starts."""
return self._start_x
@start_x.setter
def start_x(self, start_x):
try:
self._start_x = float(start_x)
except ValueError:
logger.error("Could not convert %s to a float", start_x)
raise SyntaxError(f"Could not convert {start_x} to a float")
@property
def end_x(self):
"""The x value where the evalution of the function ends."""
return self._end_x
@end_x.setter
def end_x(self, end_x):
try:
self._end_x = float(end_x)
except ValueError:
logger.error("Could not convert %s to a float", end_x)
raise SyntaxError(f"Could not convert {end_x} to a float")
class Parameter:
"""A parameter of a function.
This can be for example the standard deviation of a Gaussian function.
Args:
name (str): The name of the parameter.
symbol (str): The symbol of the parameter.
value (float): The value of the parameter.
Attributes:
name (str): The name of the parameter.
symbol (str): The symbol of the parameter.
value (float): The value of the parameter.
default (float): The default value of the parameter.
"""
def __init__(self, name: str, symbol: str, value: float) -> None:
"""Initializes the parameter."""
self.name = name
self.symbol = symbol
self.value = value
self.default = value
def set_value(self, value: float) -> None:
"""Sets the value of the parameter.
Args:
value (float): The new value of the parameter.
"""
self.value = value
logger.debug("Parameter %s set to %s", self.name, self.value)
def to_json(self) -> dict:
"""Returns a json representation of the parameter.
Returns:
dict: The json representation of the parameter.
"""
return {
"name": self.name,
"symbol": self.symbol,
"value": self.value,
"default": self.default,
}
@classmethod
def from_json(cls, data):
"""Creates a parameter from a json representation.
Args:
data (dict): The json representation of the parameter.
Returns:
Function.Parameter: The parameter.
"""
obj = cls(data["name"], data["symbol"], data["value"])
obj.default = data["default"]
return obj
class RectFunction(Function):
"""The rectangular function."""
name = "Rectangular"
def __init__(self) -> None:
"""Initializes the RecFunction."""
expr = sympy.sympify("1")
super().__init__(expr)
class SincFunction(Function):
"""The sinc function.
The sinc function is defined as sin(x * l) / (x * l).
The parameter is the scale factor l.
"""
name = "Sinc"
def __init__(self) -> None:
"""Initializes the SincFunction."""
expr = sympy.sympify("sin(x * l)/ (x * l)")
super().__init__(expr)
self.add_parameter(Function.Parameter("Scale Factor", "l", 2))
self.start_x = -np.pi
self.end_x = np.pi
class GaussianFunction(Function):
"""The Gaussian function.
The Gaussian function is defined as exp(-0.5 * ((x - mu) / sigma)**2).
The parameters are the mean and the standard deviation.
"""
name = "Gaussian"
def __init__(self) -> None:
"""Initializes the GaussianFunction."""
expr = sympy.sympify("exp(-0.5 * ((x - mu) / sigma)**2)")
super().__init__(expr)
self.add_parameter(Function.Parameter("Mean", "mu", 0))
self.add_parameter(Function.Parameter("Standard Deviation", "sigma", 1))
self.start_x = -np.pi
self.end_x = np.pi
class CustomFunction(Function):
"""A custom function."""
name = "Custom"
def __init__(self) -> None:
"""Initializes the Custom Function."""
expr = sympy.sympify(" 2 * x**2 + 3 * x + 1")
super().__init__(expr)

51
src/quackseq/helpers.py Normal file
View file

@ -0,0 +1,51 @@
"""Helper used for unit conversion."""
import decimal
class UnitConverter:
"""This class provides methods for unit conversion."""
@classmethod
def to_decimal(cls, value: str) -> decimal.Decimal:
"""This method checks if the last character of the string is a suffix.
The available suffixes are:
- n for nano
- u for micro
- m for milli
Args:
value (str): The value to be converted
Returns:
decimal.Decimal: The converted value
"""
try:
if value[-1] == "n":
return decimal.Decimal(value[:-1]) * decimal.Decimal("1e-9")
elif value[-1] == "u":
return decimal.Decimal(value[:-1]) * decimal.Decimal("1e-6")
elif value[-1] == "m":
return decimal.Decimal(value[:-1]) * decimal.Decimal("1e-3")
else:
return decimal.Decimal(value)
except TypeError:
return decimal.Decimal(value)
@classmethod
def to_float(cls, value: str) -> float:
"""This method checks if the last character of the string is a suffix.
The available suffixes are:
- n for nano
- u for micro
- m for milli
Args:
value (str): The value to be converted
Returns:
float: The converted value
"""
return float(cls.to_decimal(value))

364
src/quackseq/measurement.py Normal file
View file

@ -0,0 +1,364 @@
"""This module defines the measurement data structure and the fit class for measurement data."""
import logging
import numpy as np
from scipy.optimize import curve_fit
from quackseq.functions import Function
from quackseq.signalprocessing import SignalProcessing as sp
logger = logging.getLogger(__name__)
class Measurement:
"""This class defines how measurement data should look.
It includes pulse parameters necessary for further signal processing.
Every spectrometer should adhere to this data structure in order to be compatible with the rest of the nqrduck.
Args:
name (str): Name of the measurement.
tdx (np.array): Time axis for the x axis of the measurement data.
tdy (np.array): Time axis for the y axis of the measurement data.
target_frequency (float): Target frequency of the measurement.
frequency_shift (float, optional): Frequency shift of the measurement. Defaults to 0.
IF_frequency (float, optional): Intermediate frequency of the measurement. Defaults to 0.
Attributes:
tdx (np.array): Time axis for the x axis of the measurement data.
tdy (np.array): Time axis for the y axis of the measurement data.
target_frequency (float): Target frequency of the measurement.
frequency_shift (float): Frequency shift of the measurement.
IF_frequency (float): Intermediate frequency of the measurement.
fdx (np.array): Frequency axis for the x axis of the measurement data.
fdy (np.array): Frequency axis for the y axis of the measurement data.
"""
def __init__(
self,
name: str,
tdx: np.array,
tdy: np.array,
target_frequency: float,
frequency_shift: float = 0,
IF_frequency: float = 0,
) -> None:
"""Initializes the measurement."""
self.name = name
self.tdx = tdx
self.tdy = tdy
self.target_frequency = target_frequency
self.frequency_shift = frequency_shift
self.IF_frequency = IF_frequency
self.fdx, self.fdy = sp.fft(tdx, tdy, frequency_shift)
self.fits = []
def apodization(self, function: Function) -> "Measurement":
"""Applies apodization to the measurement data.
Args:
function (Function): Apodization function.
Returns:
Measurement: The apodized measurement.
"""
duration = (self.tdx[-1] - self.tdx[0]) * 1e-6
resolution = duration / len(self.tdx)
logger.debug("Resolution: %s", resolution)
y_weight = function.get_pulse_amplitude(duration, resolution)
tdy_apodized = self.tdy * y_weight
apodized_measurement = Measurement(
self.name,
self.tdx,
tdy_apodized,
target_frequency=self.target_frequency,
IF_frequency=self.IF_frequency,
)
return apodized_measurement
def add_fit(self, fit: "Fit") -> None:
"""Adds a fit to the measurement.
Args:
fit (Fit): The fit to add.
"""
self.fits.append(fit)
def delete_fit(self, fit: "Fit") -> None:
"""Deletes a fit from the measurement.
Args:
fit (Fit): The fit to delete.
"""
self.fits.remove(fit)
def edit_fit_name(self, fit: "Fit", name: str) -> None:
"""Edits the name of a fit.
Args:
fit (Fit): The fit to edit.
name (str): The new name.
"""
logger.debug(f"Editing fit name to {name}.")
fit.name = name
def to_json(self) -> dict:
"""Converts the measurement to a JSON-compatible format.
Returns:
dict: The measurement in JSON-compatible format.
"""
return {
"name": self.name,
"tdx": self.tdx.tolist(),
"tdy": [[x.real, x.imag] for x in self.tdy],
"target_frequency": self.target_frequency,
"IF_frequency": self.IF_frequency,
"fits": [fit.to_json() for fit in self.fits],
}
@classmethod
def from_json(cls, json: dict) -> "Measurement":
"""Converts the JSON format to a measurement.
Args:
json (dict): The measurement in JSON-compatible format.
Returns:
Measurement: The measurement.
"""
tdy = np.array([complex(y[0], y[1]) for y in json["tdy"]])
measurement = cls(
json["name"],
np.array(json["tdx"]),
tdy,
target_frequency=json["target_frequency"],
IF_frequency=json["IF_frequency"],
)
for fit_json in json["fits"]:
measurement.add_fit(Fit.from_json(fit_json, measurement))
return measurement
# Properties for encapsulation
@property
def name(self) -> str:
"""Name of the measurement."""
return self._name
@name.setter
def name(self, value: str) -> None:
self._name = value
@property
def tdx(self) -> np.array:
"""Time domain data for the measurement (x)."""
return self._tdx
@tdx.setter
def tdx(self, value: np.array) -> None:
self._tdx = value
@property
def tdy(self) -> np.array:
"""Time domain data for the measurement (y)."""
return self._tdy
@tdy.setter
def tdy(self, value: np.array) -> None:
self._tdy = value
@property
def fdx(self) -> np.array:
"""Frequency domain data for the measurement (x)."""
return self._fdx
@fdx.setter
def fdx(self, value: np.array) -> None:
self._fdx = value
@property
def fdy(self) -> np.array:
"""Frequency domain data for the measurement (y)."""
return self._fdy
@fdy.setter
def fdy(self, value: np.array) -> None:
self._fdy = value
@property
def target_frequency(self) -> float:
"""Target frequency of the measurement."""
return self._target_frequency
@target_frequency.setter
def target_frequency(self, value: float) -> None:
self._target_frequency = value
@property
def fits(self) -> list:
"""Fits of the measurement."""
return self._fits
@fits.setter
def fits(self, value: list) -> None:
self._fits = value
class Fit:
"""The fit class for measurement data. A fit can be performed on either the frequency or time domain data.
A measurement can have multiple fits.
"""
subclasses = []
def __init_subclass__(cls, **kwargs) -> None:
"""Adds the subclass to the list of subclasses."""
super().__init_subclass__(**kwargs)
cls.subclasses.append(cls)
def __init__(self, name: str, domain: str, measurement: Measurement) -> None:
"""Initializes the fit."""
self.name = name
self.domain = domain
self.measurement = measurement
self.fit()
def fit(self) -> None:
"""Fits the measurement data and sets the fit parameters and covariance."""
if self.domain == "time":
x = self.measurement.tdx
y = self.measurement.tdy
elif self.domain == "frequency":
x = self.measurement.fdx
y = self.measurement.fdy
else:
raise ValueError("Domain not recognized.")
initial_guess = self.initial_guess()
self.parameters, self.covariance = curve_fit(
self.fit_function, x, abs(y), p0=initial_guess
)
self.x = x
self.y = self.fit_function(x, *self.parameters)
def fit_function(self, x: np.array, *parameters) -> np.array:
"""The fit function.
Args:
x (np.array): The x data.
*parameters: The fit parameters.
Returns:
np.array: The y data.
"""
raise NotImplementedError
def initial_guess(self) -> list:
"""Initial guess for the fit.
Returns:
list: The initial guess.
"""
raise NotImplementedError
def to_json(self) -> dict:
"""Converts the fit to a JSON-compatible format.
Returns:
dict: The fit in JSON-compatible format.
"""
return {
"name": self.name,
"class": self.__class__.__name__,
}
@classmethod
def from_json(cls, data: dict, measurement: Measurement) -> "Fit":
"""Converts the JSON format to a fit.
Args:
data (dict): The fit in JSON-compatible format.
measurement (Measurement): The measurement.
Returns:
Fit: The fit.
"""
for subclass in cls.subclasses:
if subclass.__name__ == data["class"]:
return subclass(name=data["name"], measurement=measurement)
raise ValueError(f"Subclass {data['class']} not found.")
@property
def x(self) -> np.array:
"""The x data of the fit."""
return self._x
@x.setter
def x(self, value: np.array) -> None:
self._x = value
@property
def y(self) -> np.array:
"""The y data of the fit."""
return self._y
@y.setter
def y(self, value: np.array) -> None:
self._y = value
class T2StarFit(Fit):
"""T2* fit for measurement data."""
def __init__(self, measurement: Measurement, name: str = "T2*") -> None:
"""Initializes the T2* fit."""
super().__init__(name, "time", measurement)
def fit(self) -> None:
"""Fits the measurement data and sets the fit parameters and covariance."""
super().fit()
self.parameters = {
"S0": self.parameters[0],
"T2Star": self.parameters[1],
"covariance": self.covariance,
}
def fit_function(self, t: np.array, S0: float, T2Star: float) -> np.array:
"""The T2* fit function used for curve fitting."""
return S0 * np.exp(-t / T2Star)
def initial_guess(self) -> list:
"""Initial guess for the T2* fit."""
return [1, 1]
class LorentzianFit(Fit):
"""Lorentzian fit for measurement data."""
def __init__(self, measurement: Measurement, name: str = "Lorentzian") -> None:
"""Initializes the Lorentzian fit."""
super().__init__(name, "frequency", measurement)
def fit(self) -> None:
"""Fits the measurement data and sets the fit parameters and covariance."""
super().fit()
self.parameters = {
"S0": self.parameters[0],
"T2Star": self.parameters[1],
"covariance": self.covariance,
}
logger.debug("Lorentzian fit parameters: %s", self.parameters)
def fit_function(self, f: np.array, S0: float, T2Star: float) -> np.array:
"""The Lorentzian fit function used for curve fitting."""
return S0 / (1 + (2 * np.pi * f * T2Star) ** 2)
def initial_guess(self) -> list:
"""Initial guess for the Lorentzian fit."""
return [1, 1]

229
src/quackseq/options.py Normal file
View file

@ -0,0 +1,229 @@
import logging
from quackseq.functions import Function
logger = logging.getLogger(__name__)
class Option:
"""Defines options for the pulse parameters which can then be set accordingly.
Options can be of different types, for example boolean, numeric or function.
Args:
name (str): The name of the option.
value: The value of the option.
Attributes:
name (str): The name of the option.
value: The value of the option.
"""
subclasses = []
def __init_subclass__(cls, **kwargs):
"""Adds the subclass to the list of subclasses."""
super().__init_subclass__(**kwargs)
cls.subclasses.append(cls)
def __init__(self, name: str, value) -> None:
"""Initializes the option."""
self.name = name
self.value = value
def set_value(self):
"""Sets the value of the option.
This method has to be implemented in the derived classes.
"""
raise NotImplementedError
def to_json(self):
"""Returns a json representation of the option.
Returns:
dict: The json representation of the option.
"""
return {
"name": self.name,
"value": self.value,
"class": self.__class__.__name__,
}
@classmethod
def from_json(cls, data) -> "Option":
"""Creates an option from a json representation.
Args:
data (dict): The json representation of the option.
Returns:
Option: The option.
"""
for subclass in cls.subclasses:
logger.debug(f"Keys data: {data.keys()}")
if subclass.__name__ == data["class"]:
cls = subclass
break
# Check if from_json is implemented for the subclass
if cls.from_json.__func__ == Option.from_json.__func__:
obj = cls(data["name"], data["value"])
else:
obj = cls.from_json(data)
return obj
class BooleanOption(Option):
"""Defines a boolean option for a pulse parameter option."""
def set_value(self, value):
"""Sets the value of the option."""
self.value = value
self.value_changed.emit()
class NumericOption(Option):
"""Defines a numeric option for a pulse parameter option."""
def __init__(
self,
name: str,
value,
is_float=True,
min_value=None,
max_value=None,
slider=False,
) -> None:
"""Initializes the NumericOption.
Args:
name (str): The name of the option.
value: The value of the option.
is_float (bool): If the value is a float.
min_value: The minimum value of the option.
max_value: The maximum value of the option.
"""
super().__init__(name, value)
self.is_float = is_float
self.min_value = min_value
self.max_value = max_value
def set_value(self, value):
"""Sets the value of the option."""
if value < self.min_value:
self.value = self.min_value
self.value_changed.emit()
elif value >= self.max_value:
self.value = self.max_value
self.value_changed.emit()
else:
raise ValueError(
f"Value {value} is not in the range of {self.min_value} to {self.max_value}. This should have been caught earlier."
)
def to_json(self):
"""Returns a json representation of the option.
Returns:
dict: The json representation of the option.
"""
return {
"name": self.name,
"value": self.value,
"class": self.__class__.__name__,
"is_float": self.is_float,
"min_value": self.min_value,
"max_value": self.max_value,
}
@classmethod
def from_json(cls, data):
"""Creates a NumericOption from a json representation.
Args:
data (dict): The json representation of the NumericOption.
Returns:
NumericOption: The NumericOption.
"""
obj = cls(
data["name"],
data["value"],
is_float=data["is_float"],
min_value=data["min_value"],
max_value=data["max_value"],
)
return obj
class FunctionOption(Option):
"""Defines a selection option for a pulse parameter option.
It takes different function objects.
Args:
name (str): The name of the option.
functions (list): The functions that can be selected.
Attributes:
name (str): The name of the option.
functions (list): The functions that can be selected.
"""
def __init__(self, name, functions) -> None:
"""Initializes the FunctionOption."""
super().__init__(name, functions[0])
self.functions = functions
def set_value(self, value):
"""Sets the value of the option.
Args:
value: The value of the option.
"""
self.value = value
self.value_changed.emit()
def get_function_by_name(self, name):
"""Returns the function with the given name.
Args:
name (str): The name of the function.
Returns:
Function: The function with the given name.
"""
for function in self.functions:
if function.name == name:
return function
raise ValueError(f"Function with name {name} not found")
def to_json(self):
"""Returns a json representation of the option.
Returns:
dict: The json representation of the option.
"""
return {
"name": self.name,
"value": self.value.to_json(),
"class": self.__class__.__name__,
"functions": [function.to_json() for function in self.functions],
}
@classmethod
def from_json(cls, data):
"""Creates a FunctionOption from a json representation.
Args:
data (dict): The json representation of the FunctionOption.
Returns:
FunctionOption: The FunctionOption.
"""
logger.debug(f"Data: {data}")
# These are all available functions
functions = [Function.from_json(function) for function in data["functions"]]
obj = cls(data["name"], functions)
obj.value = Function.from_json(data["value"])
return obj

165
src/quackseq/pulseparameters.py Normal file
View file

@ -0,0 +1,165 @@
"""Contains the classes for the pulse parameters of the spectrometer. It includes the functions and the options for the pulse parameters.
Todo:
* This shouldn't be in the spectrometer module. It should be in it"s own pulse sequence module.
"""
from __future__ import annotations
import logging
from numpy.core.multiarray import array as array
from quackseq.options import BooleanOption, FunctionOption, NumericOption, Option
from quackseq.functions import RectFunction, SincFunction, GaussianFunction, CustomFunction
logger = logging.getLogger(__name__)
class PulseParameter:
"""A pulse parameter is a value that can be different for each event in a pulse sequence.
E.g. the transmit pulse power or the phase of the transmit pulse.
Args:
name (str) : The name of the pulse parameter
Attributes:
name (str) : The name of the pulse parameter
options (OrderedDict) : The options of the pulse parameter
"""
def __init__(self, name: str):
"""Initializes the pulse parameter.
Arguments:
name (str) : The name of the pulse parameter
"""
self.name = name
self.options = list()
def add_option(self, option: Option) -> None:
"""Adds an option to the pulse parameter.
Args:
option (Option) : The option to add
"""
self.options.append(option)
def get_options(self) -> list:
"""Gets the options of the pulse parameter.
Returns:
list : The options of the pulse parameter
"""
return self.options
def get_option_by_name(self, name: str) -> Option:
"""Gets an option by its name.
Args:
name (str) : The name of the option
Returns:
Option : The option with the specified name
Raises:
ValueError : If no option with the specified name is found
"""
for option in self.options:
if option.name == name:
return option
raise ValueError(f"Option with name {name} not found")
class TXPulse(PulseParameter):
"""Basic TX Pulse Parameter. It includes options for the relative amplitude, the phase and the pulse shape.
Args:
name (str): The name of the pulse parameter.
"""
RELATIVE_AMPLITUDE = "Relative TX Amplitude (%)"
TX_PHASE = "TX Phase"
TX_PULSE_SHAPE = "TX Pulse Shape"
N_PHASE_CYCLES = "Number of Phase Cycles"
PHASE_CYCLE_LEVEL = "Phase Cycle Level"
def __init__(self, name: str) -> None:
"""Initializes the TX Pulse Parameter.
It adds the options for the relative amplitude, the phase and the pulse shape.
"""
super().__init__(name)
self.add_option(
NumericOption(
self.RELATIVE_AMPLITUDE,
0,
is_float=False,
min_value=0,
max_value=100,
slider=True,
)
)
self.add_option(NumericOption(self.TX_PHASE, 0))
self.add_option(
NumericOption(
self.N_PHASE_CYCLES, 1, is_float=False, min_value=1, max_value=360
)
)
self.add_option(
NumericOption(
self.PHASE_CYCLE_LEVEL, 0, is_float=False, min_value=0, max_value=10
)
)
self.add_option(
FunctionOption(
self.TX_PULSE_SHAPE,
[
RectFunction(),
SincFunction(),
GaussianFunction(),
CustomFunction(),
],
),
)
class RXReadout(PulseParameter):
"""Basic PulseParameter for the RX Readout. It includes an option for the RX Readout state.
Args:
name (str): The name of the pulse parameter.
Attributes:
RX (str): The RX Readout state.
"""
RX = "RX"
def __init__(self, name) -> None:
"""Initializes the RX Readout PulseParameter.
It adds an option for the RX Readout state.
"""
super().__init__(name)
self.add_option(BooleanOption(self.RX, False))
class Gate(PulseParameter):
"""Basic PulseParameter for the Gate. It includes an option for the Gate state.
Args:
name (str): The name of the pulse parameter.
Attributes:
GATE_STATE (str): The Gate state.
"""
GATE_STATE = "Gate State"
def __init__(self, name) -> None:
"""Initializes the Gate PulseParameter.
It adds an option for the Gate state.
"""
super().__init__(name)
self.add_option(BooleanOption(self.GATE_STATE, False))

165
src/quackseq/pulsesequence.py Normal file
View file

@ -0,0 +1,165 @@
"""Contains the PulseSequence class that is used to store a pulse sequence and its events."""
import logging
import importlib.metadata
logger = logging.getLogger(__name__)
class PulseSequence:
"""A pulse sequence is a collection of events that are executed in a certain order.
Args:
name (str): The name of the pulse sequence
Attributes:
name (str): The name of the pulse sequence
events (list): The events of the pulse sequence
"""
def __init__(self, name, version = None) -> None:
"""Initializes the pulse sequence."""
self.name = name
# Saving version to check for compatability of saved sequence
if version is not None:
self.version = version
else:
self.version = importlib.metadata.version("nqrduck_spectrometer")
self.events = list()
def get_event_names(self) -> list:
"""Returns a list of the names of the events in the pulse sequence.
Returns:
list: The names of the events
"""
return [event.name for event in self.events]
def add_event(self, event_name: str, duration: float) -> None:
"""Add a new event to the pulse sequence.
Args:
event_name (str): The name of the event
duration (float): The duration of the event
"""
self.events.append(self.Event(event_name, f"{float(duration):.16g}u"))
def delete_event(self, event_name: str) -> None:
"""Deletes an event from the pulse sequence.
Args:
event_name (str): The name of the event to delete
"""
for event in self.events:
if event.name == event_name:
self.events.remove(event)
break
def to_json(self):
"""Returns a dict with all the data in the pulse sequence.
Returns:
dict: The dict with the sequence data
"""
# Get the versions of this package
data = {"name": self.name, "version" : self.version, "events": []}
for event in self.events:
event_data = {
"name": event.name,
"duration": event.duration,
"parameters": [],
}
for parameter in event.parameters.keys():
event_data["parameters"].append({"name": parameter, "value": []})
for option in event.parameters[parameter].options:
event_data["parameters"][-1]["value"].append(option.to_json())
data["events"].append(event_data)
return data
@classmethod
def load_sequence(cls, sequence, pulse_parameter_options):
"""Loads a pulse sequence from a dict.
The pulse paramter options are needed to load the parameters
and make sure the correct spectrometer is active.
Args:
sequence (dict): The dict with the sequence data
pulse_parameter_options (dict): The dict with the pulse parameter options
Returns:
PulseSequence: The loaded pulse sequence
Raises:
KeyError: If the pulse parameter options are not the same as the ones in the pulse sequence
"""
try:
obj = cls(sequence["name"], version = sequence["version"])
except KeyError:
logger.error("Pulse sequence version not found")
raise KeyError("Pulse sequence version not found")
for event_data in sequence["events"]:
obj.events.append(cls.Event.load_event(event_data, pulse_parameter_options))
return obj
class Variable:
"""A variable is a parameter that can be used within a pulsesequence as a placeholder.
For example the event duration a Variable with name a can be set. This variable can then be set to a list of different values.
On execution of the pulse sequence the event duration will be set to the first value in the list.
Then the pulse sequence will be executed with the second value of the list. This is repeated until the pulse sequence has
been executed with all values in the list.
"""
@property
def name(self):
"""The name of the variable."""
return self._name
@name.setter
def name(self, name: str):
if not isinstance(name, str):
raise TypeError("Name needs to be a string")
self._name = name
@property
def values(self):
"""The values of the variable. This is a list of values that the variable can take."""
return self._values
@values.setter
def values(self, values: list):
if not isinstance(values, list):
raise TypeError("Values needs to be a list")
self._values = values
class VariableGroup:
"""Variables can be grouped together.
If we have groups a and b the pulse sequence will be executed for all combinations of variables in a and b.
"""
@property
def name(self):
"""The name of the variable group."""
return self._name
@name.setter
def name(self, name: str):
if not isinstance(name, str):
raise TypeError("Name needs to be a string")
self._name = name
@property
def variables(self):
"""The variables in the group. This is a list of variables."""
return self._variables
@variables.setter
def variables(self, variables: list):
if not isinstance(variables, list):
raise TypeError("Variables needs to be a list")
self._variables = variables

86
src/quackseq/signalprocessing.py Normal file
View file

@ -0,0 +1,86 @@
"""Helper used for signal processing."""
import logging
from scipy.fft import fft, fftfreq, fftshift
import numpy as np
import sympy
logger = logging.getLogger(__name__)
class SignalProcessing:
"""This class provides various signal processing methods that can then be used by nqrduck modules."""
@classmethod
def fft(cls, tdx : np.array, tdy: np.array, freq_shift : float = 0, zero_padding = 1000) -> tuple[np.array, np.array]:
"""This method calculates the FFT of the time domain data.
Args:
tdx (np.array): Time domain x data in seconds.
tdy (np.array): Time domain magnitude y data.
freq_shift (float): Frequency shift in MHz - this can be useful if the spectrometer has it's frequency data in the IF band.
zero_padding (float): Zero padding to be used in the FFT.
Returns:
np.array: Frequency domain x data in MHz.
np.array: Frequency domain magnitude y data.
"""
dwell_time = (tdx[1] - tdx[0])
N = len(tdx) + zero_padding
if freq_shift != 0:
# Create the complex exponential to shift the frequency
shift_signal = np.exp(-2j * np.pi * freq_shift * tdx)[:, np.newaxis]
# Apply the shift by multiplying the time domain signal
tdy_shift = np.abs(tdy * shift_signal)
ydf = fftshift(fft(tdy_shift, N, axis=0), axes=0)
else:
ydf = fftshift(fft(tdy, N, axis=0), axes=0)
xdf = fftshift(fftfreq(N, dwell_time))
return xdf, ydf
@classmethod
def baseline_correction(cls, fdx : np.array, fdy : np.array, order : int) -> np.array:
"""This method calculates the baseline correction of the frequency domain data.
Args:
fdx (np.array): Frequency domain x data in MHz.
fdy (np.array): Frequency domain magnitude y data.
order (int): Order of the polynomial used for baseline correction.
Returns:
np.array: Frequency domain magnitude y data with baseline correction.
"""
pass
@classmethod
def apodization(cls, tdx : np.array, tdy : np.array, apodization_function : sympy.Expr) -> np.array:
"""This method calculates the apodization of the time domain data.
Args:
tdx (np.array): Time domain x data in seconds.
tdy (np.array): Time domain magnitude y data.
apodization_function (sympy.Expr): Apodization function.
Returns:
np.array: Time domain magnitude y data with apodization.
"""
weight = np.array([apodization_function.subs("t", t) for t in tdx])
return tdy * weight
@classmethod
def peak_picking(cls, fdx: np.array, fdy: np.array, threshold : float = 0.05) -> tuple[np.array, np.array]:
"""This method calculates the peak picking of the frequency domain data.
Args:
fdx (np.array): Frequency domain x data in MHz.
fdy (np.array): Frequency domain magnitude y data.
threshold (float): Threshold for peak picking.
Returns:
list: x,y data of the peaks.
"""
pass

92
src/quackseq/spectrometer/spectrometer_model.py Normal file
View file

@ -0,0 +1,92 @@
"""The base class for all spectrometer models."""
import logging
from collections import OrderedDict
from quackseq.spectrometer.spectrometer_setting import Setting
logger = logging.getLogger(__name__)
class SpectrometerModel():
"""The base class for all spectrometer models.
It contains the settings and pulse parameters of the spectrometer.
Args:
module (Module) : The module that the spectrometer is connected to
Attributes:
settings (OrderedDict) : The settings of the spectrometer
pulse_parameter_options (OrderedDict) : The pulse parameter options of the spectrometer
"""
settings: OrderedDict
pulse_parameter_options: OrderedDict
def __init__(self, module):
"""Initializes the spectrometer model.
Args:
module (Module) : The module that the spectrometer is connected to
"""
super().__init__(module)
self.settings = OrderedDict()
self.pulse_parameter_options = OrderedDict()
def add_setting(self, setting: Setting, category: str) -> None:
"""Adds a setting to the spectrometer.
Args:
setting (Setting) : The setting to add
category (str) : The category of the setting
"""
if category not in self.settings.keys():
self.settings[category] = []
self.settings[category].append(setting)
def get_setting_by_name(self, name: str) -> Setting:
"""Gets a setting by its name.
Args:
name (str) : The name of the setting
Returns:
Setting : The setting with the specified name
Raises:
ValueError : If no setting with the specified name is found
"""
for category in self.settings.keys():
for setting in self.settings[category]:
if setting.name == name:
return setting
raise ValueError(f"Setting with name {name} not found")
def add_pulse_parameter_option(
self, name: str, pulse_parameter_class: "PulseParameter"
) -> None:
"""Adds a pulse parameter option to the spectrometer.
Args:
name (str) : The name of the pulse parameter
pulse_parameter_class (PulseParameter) : The pulse parameter class
"""
self.pulse_parameter_options[name] = pulse_parameter_class
@property
def target_frequency(self):
"""The target frequency of the spectrometer in Hz. This is the frequency where the magnetic resonance experiment is performed."""
raise NotImplementedError
@target_frequency.setter
def target_frequency(self, value):
raise NotImplementedError
@property
def averages(self):
"""The number of averages for the spectrometer."""
raise NotImplementedError
@averages.setter
def averages(self, value):
raise NotImplementedError

243
src/quackseq/spectrometer/spectrometer_setting.py Normal file
View file

@ -0,0 +1,243 @@
"""Settings for the different spectrometers."""
import logging
logger = logging.getLogger(__name__)
class Setting():
"""A setting for the spectrometer is a value that is the same for all events in a pulse sequence.
E.g. the Transmit gain or the number of points in a spectrum.
Args:
name (str) : The name of the setting
description (str) : A description of the setting
default : The default value of the setting
Attributes:
name (str) : The name of the setting
description (str) : A description of the setting
value : The value of the setting
"""
def __init__(self, name: str, description: str = None, default=None) -> None:
"""Create a new setting.
Args:
name (str): The name of the setting.
description (str): A description of the setting.
default: The default value of the setting.
"""
self.widget = None
super().__init__()
self.name = name
self.description = description
if default is not None:
self.value = default
# Update the description with the default value
self.description += f"\n (Default: {default})"
class NumericalSetting(Setting):
"""A setting that is a numerical value.
It can additionally have a minimum and maximum value.
"""
def __init__(
self, name: str, description: str, default, min_value=None, max_value=None
) -> None:
"""Create a new numerical setting."""
super().__init__(
name,
self.description_limit_info(description, min_value, max_value),
default,
)
def description_limit_info(self, description: str, min_value, max_value) -> str:
"""Updates the description with the limits of the setting if there are any.
Args:
description (str): The description of the setting.
min_value: The minimum value of the setting.
max_value: The maximum value of the setting.
Returns:
str: The description of the setting with the limits.
"""
if min_value is not None and max_value is not None:
description += f"\n (min: {min_value}, max: {max_value})"
elif min_value is not None:
description += f"\n (min: {min_value})"
elif max_value is not None:
description += f"\n (max: {max_value})"
return description
class FloatSetting(NumericalSetting):
"""A setting that is a Float.
Args:
name (str) : The name of the setting
default : The default value of the setting
description (str) : A description of the setting
min_value : The minimum value of the setting
max_value : The maximum value of the setting
"""
DEFAULT_LENGTH = 100
def __init__(
self,
name: str,
default: float,
description: str,
min_value: float = None,
max_value: float = None,
) -> None:
"""Create a new float setting."""
super().__init__(name, description, default, min_value, max_value)
@property
def value(self):
"""The value of the setting. In this case, a float."""
return self._value
@value.setter
def value(self, value):
logger.debug(f"Setting {self.name} to {value}")
self._value = float(value)
self.settings_changed.emit()
class IntSetting(NumericalSetting):
"""A setting that is an Integer.
Args:
name (str) : The name of the setting
default : The default value of the setting
description (str) : A description of the setting
min_value : The minimum value of the setting
max_value : The maximum value of the setting
"""
def __init__(
self,
name: str,
default: int,
description: str,
min_value=None,
max_value=None,
spin_box: tuple = (False, False),
) -> None:
"""Create a new int setting."""
self.spin_box = spin_box
super().__init__(name, description, default, min_value, max_value)
@property
def value(self):
"""The value of the setting. In this case, an int."""
return self._value
@value.setter
def value(self, value):
logger.debug(f"Setting {self.name} to {value}")
value = int(float(value))
self._value = value
class BooleanSetting(Setting):
"""A setting that is a Boolean.
Args:
name (str) : The name of the setting
default : The default value of the setting
description (str) : A description of the setting
"""
def __init__(self, name: str, default: bool, description: str) -> None:
"""Create a new boolean setting."""
super().__init__(name, description, default)
@property
def value(self):
"""The value of the setting. In this case, a bool."""
return self._value
@value.setter
def value(self, value):
try:
self._value = bool(value)
except ValueError:
raise ValueError("Value must be a bool")
class SelectionSetting(Setting):
"""A setting that is a selection from a list of options.
Args:
name (str) : The name of the setting
options (list) : A list of options to choose from
default : The default value of the setting
description (str) : A description of the setting
"""
def __init__(
self, name: str, options: list, default: str, description: str
) -> None:
"""Create a new selection setting."""
super().__init__(name, description, default)
# Check if default is in options
if default not in options:
raise ValueError("Default value must be one of the options")
self.options = options
@property
def value(self):
"""The value of the setting. In this case, a string."""
return self._value
@value.setter
def value(self, value):
try:
if value in self.options:
self._value = value
else:
raise ValueError("Value must be one of the options")
# This fixes a bug when creating the widget when the options are not yet set
except AttributeError:
self._value = value
self.options = [value]
self.settings_changed.emit()
class StringSetting(Setting):
"""A setting that is a string.
Args:
name (str) : The name of the setting
default : The default value of the setting
description (str) : A description of the setting
"""
def __init__(self, name: str, default: str, description: str) -> None:
"""Create a new string setting."""
super().__init__(name, description, default)
@property
def value(self):
"""The value of the setting. In this case, a string."""
return self._value
@value.setter
def value(self, value):
try:
self._value = str(value)
except ValueError:
raise ValueError("Value must be a string")

38
tests/simulator.py Normal file
View file

@ -0,0 +1,38 @@
# Dummy test to communicate the structure
from quackseq.pulsesequence import PulseSequence
from quackseq.event import Event
from quackseq.functions import RectFunction
seq = PulseSequence("test")
tx = Event("tx", "10u")
# tx.set_tx_amplitude(1)
#tx.set_tx_phase(0)
#tx.set_shape(RectFunction())
#seq.add_event(tx)
#blank = Event("blank", "10u")
#seq.add_event(blank)
#rx = Event("rx", "10u")
#rx.set_rx_phase(0)
#rx.set_rx(True)
#seq.add_event(rx)
#TR = Event("TR", "1ms")
#seq.add_event(TR)
#sim = Simulator()
#sim.set_averages(100)
# Returns the data at the RX event
#result = sim.run(seq)
#result.plot()