import numpy as np
import pandas as pd
import skrf as rf
from qdrive.dataset import dataset
from scipy.optimize import curve_fit
from semi_cr.core.lab.utils.post_processing_tools.base import BaseAnalyzer
from semi_cr.core.lab.utils.post_processing_tools.dataqruiser_utils import get_dataset_by_uuid
from semi_cr.core.lab.utils.post_processing_tools.vna_utils import load_vna_csv
[docs]
class SParamAnalyzer(BaseAnalyzer):
"""
Class for analyzing S-parameters.
"""
def __init__(self):
super().__init__()
[docs]
def load_from_uuid(self, uuid):
"""
Load the qcodes dataset given its UUID.
"""
ds: dataset = get_dataset_by_uuid(uuid)
return ds
[docs]
def align_axis(self, axis, data, target_axis):
"""
Align the given axis of the data to the target axis.
Args:
axis: Original axis.
data: Original data.
target_axis: Target axis to align to.
Returns:
target_data: interpolated data aligned to the target axis.
"""
aligned_data = np.interp(target_axis, axis, data)
return aligned_data
[docs]
def delftacDataset_to_dict(self, qcodes_dataset: dataset) -> tuple[dict[str, pd.DataFrame], dict[str, np.ndarray]]:
"""
Convert a QCoDeS dataset to a dictionary of pandas DataFrames for S parameters.
Args:
qcodes_dataset: The QCoDeS dataset to convert.
Returns:
A tuple containing:
- A dictionary where the key represent the connection_info and the value is the corresponding xarray dataset.
- A dictionary with coordinates as numpy arrays.
"""
coordinates = qcodes_dataset[next(iter(qcodes_dataset.files.keys()))].xarray.coords
dataset_dict = {}
for file in qcodes_dataset.files.keys():
attributes = file.split("_")
# subjectID = attributes[0]
# type_of_measurement = attributes[1]
connection_info = attributes[2]
# RunID = attributes[3]
xr = qcodes_dataset[file].xarray
dataset_dict[connection_info] = xr
coordinates = {name: coord.to_numpy() for name, coord in coordinates.items()}
return dataset_dict, coordinates
[docs]
def sampleboardDataset_to_dict(self, qcodes_dataset: dataset) -> tuple[dict[str, pd.DataFrame], dict[str, np.ndarray]]:
"""
For sample board baseline data: Convert a QCoDeS dataset to a dictionary of pandas DataFrames for S parameters.
This function is for load the sample board baseline data, which consists of transmission and crosstalk. This is shown as frequency in Hz vs S amplitude in dB
Args:
qcodes_dataset: The QCoDeS dataset to convert.
Returns:
A tuple containing:
- A dictionary where the key represent the connection_info and the value is the corresponding xarray dataset.
- A dictionary with coordinates as numpy arrays.
"""
coordinates = qcodes_dataset[next(iter(qcodes_dataset.files.keys()))].xarray.coords
dataset_dict = {}
for file in qcodes_dataset.files.keys():
attributes = file.split("_")
# subjectID = attributes[0]
# type_of_measurement = attributes[1]
connection_info = attributes[2]
# RunID = attributes[3]
xr = qcodes_dataset[file].xarray
dataset_dict[connection_info] = xr
coordinates = {name: coord.to_numpy() for name, coord in coordinates.items()}
return dataset_dict, coordinates
[docs]
def muxQcodesDataset_to_dict_V2(self, qcodes_dataset: dataset, filter: dict[str, float | tuple[float, float] | str] | None = None) -> tuple[dict[str, pd.DataFrame], dict[str, np.ndarray]]:
"""
Convert a QCoDeS dataset to a dictionary of pandas DataFrames for S parameters.
Args:
qcodes_dataset: The QCoDeS dataset to convert.
filter: Optional dictionary specifying the filter criteria for the dataset. The keys are the names of the coordinates, and the values are either a single value or a range (tuple) to filter on.
"""
mux_dataset_dict = {}
print(qcodes_dataset.files.keys())
coordinates = qcodes_dataset[next(iter(qcodes_dataset.files.keys()))].xarray.coords
for file in qcodes_dataset.files.keys():
channel_name = file.split("_")[0]
if channel_name not in {"channel1", "channel2", "channel3", "channel4"}:
continue # Skip files that do not correspond to the expected channel names
xr = qcodes_dataset[file].xarray
if channel_name == "channel1":
R = xr.R_ch1
elif channel_name == "channel2":
R = xr.R_ch2
elif channel_name == "channel3":
R = xr.R_ch3
elif channel_name == "channel4":
R = xr.R_ch4
mux_dataset_dict[channel_name] = R.to_numpy()
coordinates = {name: coord.to_numpy() for name, coord in coordinates.items()}
filtered_mux_dataset_dict = {}
if filter is not None:
names = list(filter.keys())
value_ranges = list(filter.values())
masks = {}
for name, value_range in zip(names, value_ranges):
if name in coordinates:
idx = list(coordinates.keys()).index(name)
if value_range == "all":
continue
elif isinstance(value_range, tuple):
# locate which indices of the coordinate values are within the specified range
min_value, max_value = value_range
mask = ((coordinates[name] >= min_value) & (coordinates[name] <= max_value))
coordinates[name] = coordinates[name][mask]
masks[idx] = mask
else:
# means only need single value, int or float, or str
mask = (coordinates[name] == value_range)
coordinates[name] = coordinates[name][mask]
masks[idx] = mask
else:
raise ValueError(f"Coordinate '{name}' not found in the dataset coordinates: {list(coordinates.keys())}.")
# get total mask
index = [slice(None)] * mux_dataset_dict[next(iter(mux_dataset_dict))].ndim
for idx, mask in masks.items():
index[idx] = mask
# apply total mask to data
for channel_name, data in mux_dataset_dict.items():
filtered_mux_dataset_dict[channel_name] = data[tuple(index)]
return filtered_mux_dataset_dict, coordinates
[docs]
def muxQcodesDataset_to_dict(self, qcodes_dataset, filter=None) -> tuple[dict[str, pd.DataFrame], dict[str, np.ndarray]]:
"""
Convert a QCoDeS dataset to a dictionary of pandas DataFrames for S parameters.
Args:
qcodes_dataset: The QCoDeS dataset to convert.
filter: Optional dictionary specifying the filter criteria for the dataset. The keys are the names of the coordinates, and the values are either a single value or a range (tuple) to filter on.
Returns:
A tuple containing:
- A dictionary with S parameters as pandas DataFrames.
- A dictionary with coordinates as numpy arrays.
"""
mux_dataset_dict = {}
print(qcodes_dataset.files.keys())
coordinates = qcodes_dataset[next(iter(qcodes_dataset.files.keys()))].xarray.coords
for file in qcodes_dataset.files.keys():
channel_name = file.split("_")[1]
xr = qcodes_dataset[file].xarray
if channel_name == "channel1":
R = xr.R_ch1
elif channel_name == "channel2":
R = xr.R_ch2
elif channel_name == "channel3":
R = xr.R_ch3
elif channel_name == "channel4":
R = xr.R_ch4
mux_dataset_dict[channel_name] = R.to_numpy()
coordinates = {name: coord.to_numpy() for name, coord in coordinates.items()}
filtered_mux_dataset_dict = {}
if filter is not None:
names = list(filter.keys())
value_ranges = list(filter.values())
masks = {}
for name, value_range in zip(names, value_ranges):
if name in coordinates:
idx = list(coordinates.keys()).index(name)
if value_range == "all":
continue
elif isinstance(value_range, tuple):
# locate which indices of the coordinate values are within the specified range
min_value, max_value = value_range
mask = ((coordinates[name] >= min_value) & (coordinates[name] <= max_value))
coordinates[name] = coordinates[name][mask]
masks[idx] = mask
else:
# means only need single value, int or float, or str
mask = (coordinates[name] == value_range)
coordinates[name] = coordinates[name][mask]
masks[idx] = mask
else:
raise ValueError(f"Coordinate '{name}' not found in the dataset coordinates: {list(coordinates.keys())}.")
# get total mask
index = [slice(None)] * mux_dataset_dict[next(iter(mux_dataset_dict))].ndim
for idx, mask in masks.items():
index[idx] = mask
# apply total mask to data
for channel_name, data in mux_dataset_dict.items():
filtered_mux_dataset_dict[channel_name] = data[tuple(index)]
return filtered_mux_dataset_dict, coordinates
[docs]
def logistic(self, x, y0, A, x0, k):
"""
Logistic function for curve fitting.
"""
return y0 + A / (1 + np.exp(-(x - x0) / k))
[docs]
def fit_rise_time(self, x, y):
"""
Fit a logistic curve and return the 10-90% rise time.
Returns:
-------
popt : tuple
(y0, A, x0, k)
rise_time : float
10-90% rise time.
"""
p0 = (
np.min(y),
np.max(y) - np.min(y),
x[np.argmax(np.gradient(y))],
(x[-1] - x[0]) / 20,
)
popt, _ = curve_fit(self.logistic, x, y, p0=p0)
_, _, _, k = popt
rise_time = 2 * np.log(9) * abs(k)
return popt, rise_time
[docs]
def absoluteV_to_dB(self, absolute_value: float | np.ndarray | list, reference: float = 0.15) -> float | np.ndarray:
return 20 * np.log10(np.abs(absolute_value) / reference)
[docs]
def load_from_vna_csv(self, file_path) -> dict[str, pd.DataFrame]:
"""
Load the S-parameter dataset from a VNA CSV file and return S parameters as a dictionary of pandas DataFrames.
Args:
file_path (str): Path to the VNA CSV file.
"""
df = load_vna_csv(file_path)
s11 = df[["Freq_GHz", "S11"]]
s21 = df[["Freq_GHz", "S21"]]
s12 = df[["Freq_GHz", "S12"]]
s22 = df[["Freq_GHz", "S22"]]
data = {'s11': s11, 's21': s21, 's12': s12, 's22': s22}
return data
[docs]
def load_s2p_db(self, file_path: str) -> dict[str, pd.DataFrame]:
"""
Load an .s2p Touchstone file and return S-parameters in dB.
Args:
file_path: Path to the .s2p file.
Returns:
Dictionary with keys 's11', 's21', 's12', 's22'.
Each value is a DataFrame with columns:
- Freq_GHz
- Sxx (dB)
"""
ntwk = rf.Network(file_path)
freq_ghz = ntwk.f / 1e9
data = {
"s11": pd.DataFrame({
"Freq_GHz": freq_ghz,
"S11": 20 * np.log10(np.abs(ntwk.s[:, 0, 0])),
}),
"s21": pd.DataFrame({
"Freq_GHz": freq_ghz,
"S21": 20 * np.log10(np.abs(ntwk.s[:, 1, 0])),
}),
"s12": pd.DataFrame({
"Freq_GHz": freq_ghz,
"S12": 20 * np.log10(np.abs(ntwk.s[:, 0, 1])),
}),
"s22": pd.DataFrame({
"Freq_GHz": freq_ghz,
"S22": 20 * np.log10(np.abs(ntwk.s[:, 1, 1])),
}),
}
return data
[docs]
def get_dataset_attributes_from_uuid(self, uuid):
"""
Retrieve dataset attributes from a dataset given its UUID.
Args:
uuid (str): The UUID of the dataset.
"""
# Implement the logic to retrieve QHarbour parameters using the provided UUID
pass
[docs]
def get_dataset_attributes_from_yaml(self, yaml_file_path):
"""
Retrieve dataset attributes from a YAML file.
Args:
yaml_file_path (str): Path to the YAML file.
"""
# Implement the logic to retrieve QHarbour parameters using the provided YAML file
pass
[docs]
def analyze(self):
"""
Analyze the S-parameter data and return relevant metrics.
"""
pass
[docs]
def extract_3dB_bandwidth(self, frequencies: np.ndarray, insertion_loss: np.ndarray) -> float:
"""
Extract the 3dB bandwidth from the insertion loss data. This will take the minimal value of the insertion loss as the reference and find the frequencies where the insertion loss is 3dB above that reference.
Args:
frequencies (np.ndarray): Frequency data in Hz.
insertion_loss (np.ndarray): Insertion loss data in dB.
"""
min_insertion_loss = np.min(insertion_loss)
idx_3dB = np.where(insertion_loss <= min_insertion_loss + 3)[0]
if len(idx_3dB) == 0:
raise ValueError("Unable to find valid indices for 3dB bandwidth calculation.")
freq_3dB = frequencies[idx_3dB]
bandwidth = np.max(freq_3dB) - np.min(freq_3dB)
return bandwidth
[docs]
def extract_3dB_bandwidth_from_transmission(self, frequencies: np.ndarray, transmission: np.ndarray) -> float:
"""
Extract the 3dB bandwidth from the transmission data. This will take the maximal value of the transmission as the reference and find the frequencies where the transmission is 3dB below that reference.
Args:
frequencies (np.ndarray): Frequency data in Hz.
transmission (np.ndarray): Transmission data in dB.
"""
max_transmission = np.max(transmission)
idx_3dB = np.where(transmission >= max_transmission - 3)[0]
if len(idx_3dB) == 0:
raise ValueError("Unable to find valid indices for 3dB bandwidth calculation.")
freq_3dB = frequencies[idx_3dB]
bandwidth = np.max(freq_3dB) - np.min(freq_3dB)
return bandwidth
[docs]
def find_breakpoints(self, frequencies: np.ndarray, insertion_loss: np.ndarray) -> tuple[float, float]:
"""
Find the breakpoints in the insertion loss data where the slope changes significantly.
Args:
frequencies (np.ndarray): Frequency data in Hz.
insertion_loss (np.ndarray): Insertion loss data in dB.
"""
pass