Source code for semi_cr.core.lab.utils.post_processing_tools.resistanceMatrixAnalyzer

from typing import TYPE_CHECKING

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from matplotlib.colors import LogNorm

from semi_cr.core.lab.station.context import StationContext
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.setup.juno.cqcc.utils.reference_tools import get_full_DC_expected_resistance_matrix, get_full_DC_series_resistance_matrix

if TYPE_CHECKING:
    from qdrive.dataset import dataset


[docs] class ResistanceMatrixAnalyzer(BaseAnalyzer): """ Class for analyzing resistance matrix datasets. """ def __init__(self, uuid): super().__init__() self.uuid = uuid
[docs] def load(self): """ Load the dataset given its UUID. """ self.resistance_matrix_np, self.source_channels, self.drain_channels = self.get_measurement_data_as_numpy(self.uuid)
[docs] def analyze(self): """ Perform analysis on the resistance matrix dataset. This method can be extended to include specific analysis logic. """ # Example: Print the shape of the resistance matrix print(f"Resistance matrix shape: {self.resistance_matrix_np.shape}")
# Additional analysis logic can be added here
[docs] def plot_resistance_matrix(self, vmin, vmax): _, ax = plt.subplots() pcm = ax.pcolormesh( self.source_channels, self.drain_channels, self.resistance_matrix_np, norm=LogNorm(vmin=vmin, vmax=vmax) ) plt.xlim(0, 97) plt.ylim(0, 97) plt.xlabel("Source Channel") plt.ylabel("Drain Channel") plt.colorbar(pcm, label="Resistance (Ohms)") plt.show()
[docs] def get_measurement_data_as_numpy(self, uuid: str) -> tuple[np.ndarray, np.ndarray, np.ndarray]: """ Get the resistancemeasurement data from a dataset given its UUID as a format of a numpy array. Also return its coordinates Args: uuid (str): The UUID of the dataset. """ ds: dataset = get_dataset_by_uuid(uuid) xr_ds = ds[next(iter(ds.files.keys()))].xarray coordinates = xr_ds.coords source_channels = coordinates["source_channel"].values drain_channels = coordinates["drain_channel"].values resistance_matrix = xr_ds['resistance_matrix'] resistance_matrix_np = resistance_matrix.to_numpy() return resistance_matrix_np, source_channels, drain_channels
[docs] def get_reference_resistance_matrix(self, station_context: StationContext) -> pd.DataFrame: """ _summary_ ----------- Get the reference resistance matrix from the station context. This method can be extended to include specific logic for extracting the reference resistance matrix. Args: ----------- station_context (StationContext): The station context containing connector information. """ return get_full_DC_expected_resistance_matrix(station_context, 1)
[docs] def get_series_resistance_matrix(self, station_context: StationContext) -> pd.DataFrame: """ Get the series resistance matrix from the station context. This method can be extended to include specific logic for extracting the series resistance matrix. Args: station_context (StationContext): The station context containing connector information. """ return get_full_DC_series_resistance_matrix(station_context, 1)
[docs] def get_difference_log(self, reference_matrix: np.ndarray, measurement_matrix: np.ndarray) -> np.ndarray: """_summary_ Get the log difference between the reference and measurement resistance matrices. Args: reference_matrix (np.ndarray): reference resistance matrix as a numpy array. measurement_matrix (np.ndarray): measurement resistance matrix as a numpy array. Returns: np.ndarray: The log difference between the reference and measurement resistance matrices. """ diff_log = np.log10(reference_matrix) - np.log10(measurement_matrix) return diff_log
[docs] def filter_with_significance_log(self, diff_log: np.ndarray, threshold: float) -> np.ndarray: """_summary_ Filter the log difference matrix based on a significance threshold. Args: diff_log (np.ndarray): The log difference between the reference and measurement resistance matrices. threshold (float): The significance threshold for filtering. Returns: np.ndarray: The filtered log difference matrix. """ diff_log[(diff_log >= -threshold) & (diff_log <= threshold)] = 0 return diff_log