from typing import TYPE_CHECKING
import matplotlib.pyplot as plt
import networkx as nx
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 (
convert_dc_to_device_resistance_matrix,
get_full_DC_expected_resistance_matrix,
get_full_DC_series_resistance_matrix,
)
if TYPE_CHECKING:
import xarray as xr
from qdrive.dataset import dataset
def _dc_channel_label(channel: object) -> str:
"""Convert a positive integer-like channel coordinate to ``DC<number>``."""
if isinstance(channel, (bool, np.bool_)):
raise ValueError(f"Expected a positive integer channel, got {channel!r}")
try:
numeric_channel = float(channel)
except (TypeError, ValueError) as error:
raise ValueError(f"Expected a positive integer channel, got {channel!r}") from error
if not np.isfinite(numeric_channel) or not numeric_channel.is_integer() or numeric_channel < 1:
raise ValueError(f"Expected a positive integer channel, got {channel!r}")
return f"DC{int(numeric_channel)}"
def _resistance_dataarray_to_dataframe(resistance_matrix: "xr.DataArray") -> pd.DataFrame:
expected_dimensions = {"source_channel", "drain_channel"}
if set(resistance_matrix.dims) != expected_dimensions:
raise ValueError(f"Resistance matrix must have exactly the dimensions {sorted(expected_dimensions)!r}, got {list(resistance_matrix.dims)!r}")
dataframe = resistance_matrix.transpose("source_channel", "drain_channel").to_pandas()
dataframe.index = [_dc_channel_label(channel) for channel in dataframe.index]
dataframe.columns = [_dc_channel_label(channel) for channel in dataframe.columns]
if not dataframe.index.is_unique:
raise ValueError("Source channel coordinates must be unique")
if not dataframe.columns.is_unique:
raise ValueError("Drain channel coordinates must be unique")
dataframe.index.name = "source_channel"
dataframe.columns.name = "drain_channel"
return dataframe
[docs]
class ResistanceMatrixAnalyzer(BaseAnalyzer):
"""
Class for analyzing resistance matrix datasets.
"""
def __init__(self, uuid):
super().__init__()
self.uuid = uuid
self.resistance_matrix_np = None
self.resistance_matrix_pd = None
[docs]
def load(self, type="np"):
"""
Load the dataset given its UUID.
"""
self.resistance_matrix_np, self.source_channels, self.drain_channels = self.get_measurement_data_as_numpy(self.uuid)
if type == "pd":
self.resistance_matrix_pd = self.get_measurement_data_as_pdDataframe(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_measurement_data_as_pdDataframe(self, uuid: str) -> pd.DataFrame:
"""
Get resistance data as a source-by-drain pandas DataFrame.
Both axes use canonical DC labels such as ``DC1`` through ``DC96``.
Args:
uuid (str): The UUID of the dataset.
"""
ds: dataset = get_dataset_by_uuid(uuid)
xr_ds = ds[next(iter(ds.files.keys()))].xarray
return _resistance_dataarray_to_dataframe(xr_ds["resistance_matrix"])
[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
[docs]
def cqcc_station_results2report(
self,
graph: nx.MultiDiGraph,
):
"""
Convert the results from the CQCC station analysis into a report format.
Args:
graph (nx.MultiDiGraph): The graph representation of the resistance matrix.
Returns:
dict: A dictionary containing the report data.
"""
if self.resistance_matrix_pd is None:
raise ValueError("Resistance matrix DataFrame is not loaded. Please call load(type='pd') first.")
ROI_resistance_matrix, ROI_subset = convert_dc_to_device_resistance_matrix(self.resistance_matrix_pd, graph)
switch_helth_resistance_matrix = self.resistance_matrix_np.diagonal()
# now get the opencircuit subset from full matrix
opencircuit_resistance_matrix = self.resistance_matrix_pd.copy()
opencircuit_resistance_matrix.loc[ROI_subset.index, ROI_subset.columns] = np.nan
return ROI_resistance_matrix, switch_helth_resistance_matrix, opencircuit_resistance_matrix