import pandas as pd import matplotlib.pyplot as plt from flask import Blueprint, render_template_string, request, jsonify, Response import os from sklearn.decomposition import PCA from sklearn.preprocessing import StandardScaler import numpy as np import json from scipy.optimize import curve_fit from scipy.stats import linregress # Create Blueprint her_plot_bp = Blueprint('her_plot', __name__, url_prefix='/her') # Atomic weights for conversion between atomic and weight fractions ATOMIC_WEIGHTS = { 'Ag': 107.8682, 'Au': 196.966569, 'Cd': 112.411, 'Cu': 63.546, 'Ga': 69.723, 'Hg': 200.59, 'In': 114.818, 'Mn': 54.938044, 'Mo': 95.96, 'Nb': 92.90637, 'Ni': 58.6934, 'Pd': 106.42, 'Pt': 195.084, 'Rh': 102.90550, 'Sn': 118.710, 'Tl': 204.38, 'W': 183.84, 'Zn': 65.38 } # Voltage conversion functions def lin_fxn(x, a, b): return a*x+b def fit_lin(X, Y): params, covariance = curve_fit(lin_fxn, X, Y) a_fit, b_fit = params return (a_fit, b_fit) def est_x(x, X, Y): fit = fit_lin(X, Y) x = fit[0]*x+fit[1] return x def load_calibration_data_for_voltage_conversion(custom_params=None): """Load calibration data for voltage conversion from full cell to half cell.""" # Default experiment conditions: Neutral CO2RR in 4cm2 cell, Sputtered Copper Catalyst, 0.1M Bicarbonate - ref electrode (3M kcl) 230mV vs SHE default_params = { 'ref_pot': 0.23, # V Ag/AgCl electrode 'cathode_pH': 10, 'anode_pH': 3, 'geo_area': 4, # cm2 'membrane_loss': 0.1, # V # Note: anode_measured_potential_vs_ref is now interpolated from calibration data } # Use custom parameters if provided, otherwise use defaults if custom_params: params = {**default_params, **custom_params} else: params = default_params ref_pot = params['ref_pot'] cathode_pH = params['cathode_pH'] anode_pH = params['anode_pH'] Nern_pH_loss = (cathode_pH-anode_pH)*0.059 geo_area = params['geo_area'] membrane_loss = params['membrane_loss'] # Measurements from calibration work j = np.array([50,100,200]) cathode_pot = np.array([-1.62,-2.0,-2.3]) cathode_R = np.array([0.48,0.34,0.3]) anode_pot = np.array([1.3,1.35,1.4]) anode_R = np.array([0,0,0]) #almost negligible fullcell_pot = np.array([3,3.4,3.7]) fullcell_R = np.array([0.47,0.35,0.3]) n = len(cathode_pot) cathode_pot_corr = np.zeros(n) anode_pot_corr = np.zeros(n) cathode_overpot = np.zeros(n) anode_overpot = np.zeros(n) fullcell_pot_corr = np.zeros(n) for i in range(0, n): cathode_pot_corr[i] = correct_potential(cathode_pot[i], cathode_R[i], cathode_pH, j[i], geo_area, ref_pot) anode_pot_corr[i] = correct_potential(anode_pot[i], anode_R[i], anode_pH, j[i], geo_area, ref_pot) cathode_overpot[i] = get_overpotential(cathode_pot_corr[i],0.08) anode_overpot[i] = get_overpotential(anode_pot_corr[i], 1.23) fullcell_pot_corr[i] = fullcell_pot[i]-fullcell_R[i]*j[i]/1000*geo_area conditions_dict = { 'ref pot': ref_pot, 'cathode pH': cathode_pH, 'anode pH': anode_pH, 'Nern pH loss': Nern_pH_loss, 'geo area': geo_area, 'membrane loss': membrane_loss, } measurements_dict = { 'j': j, 'cathode pot': cathode_pot, 'cathode R': cathode_R, 'anode pot': anode_pot, 'anode R': anode_R, 'fullcell pot': fullcell_pot, 'fullcell R': fullcell_R, } data_dict = { 'cathode pot corr': cathode_pot_corr, 'anode pot corr': anode_pot_corr, 'cathode overpot': cathode_overpot, 'anode overpot': anode_overpot, 'fullcell pot corr': fullcell_pot_corr } return {'measurements': measurements_dict, 'conditions': conditions_dict, 'extracted params': data_dict} def she2rhe(ushe, pH, ref_pot): ushe = ushe+ref_pot+(0.059*pH) return ushe def rhe2she(urhe, pH, ref_pot): urhe = urhe - (0.059 * pH) return urhe def correct_potential(pot, R, pH, j, area, ref_pot): if pot<0: corrected_pot = she2rhe(pot+j/1000*area*R,pH, ref_pot) else: corrected_pot = she2rhe(pot-j/1000*area*R,pH, ref_pot) return corrected_pot def interpolate_cathode_R(current_density): """ Interpolate cathode resistance R from log(j) vs R calibration data. Calibration data: j = [50, 100, 200] mA/cm² R = [0.48, 0.34, 0.3] ohm Fits log(j) vs R and interpolates R for given current density. """ # Calibration data j_array = np.array([50, 100, 200]) # mA/cm² R_array = np.array([0.48, 0.34, 0.3]) # ohm # Convert to log scale for j log_j = np.log10(j_array) # Fit linear relationship: R = a * log10(j) + b fit_params = np.polyfit(log_j, R_array, 1) a, b = fit_params # Interpolate R for given current density (convert mA/cm² to mA/cm², already in correct units) if current_density <= 0: # Use minimum R if current density is too small return R_array[-1] # Use the smallest R (at highest j) log_j_input = np.log10(current_density) R_interpolated = a * log_j_input + b # Clamp to reasonable bounds (between min and max R values) R_interpolated = np.clip(R_interpolated, R_array.min(), R_array.max()) return R_interpolated def interpolate_anode_potential_vs_ref(current_density): """ Interpolate anode measured potential vs reference from log(j) vs anode_pot calibration data. Calibration data: j = [50, 100, 200] mA/cm² anode_pot = [1.3, 1.35, 1.4] V Fits log(j) vs anode_pot and interpolates anode_pot for given current density. """ # Calibration data j_array = np.array([50, 100, 200]) # mA/cm² anode_pot_array = np.array([1.3, 1.35, 1.4]) # V # Convert to log scale for j log_j = np.log10(j_array) # Fit linear relationship: anode_pot = a * log10(j) + b fit_params = np.polyfit(log_j, anode_pot_array, 1) a, b = fit_params # Interpolate anode_pot for given current density if current_density <= 0: # Use minimum anode_pot if current density is too small return anode_pot_array[0] # Use the smallest anode_pot (at lowest j) log_j_input = np.log10(current_density) anode_pot_interpolated = a * log_j_input + b # Clamp to reasonable bounds (between min and max anode_pot values) anode_pot_interpolated = np.clip(anode_pot_interpolated, anode_pot_array.min(), anode_pot_array.max()) return anode_pot_interpolated def cell2rhe(vcell, ref_pot, anode_pH, membrane_loss, Nern_pH_loss, current_density, geo_area, custom_anode_potential_vs_ref=None, custom_R_cathode=None): """ Convert full cell voltage to cathode potential vs RHE. Steps (matching notebook example): 1. Interpolate anode measured potential vs reference from calibration data (or use custom value) 2. Convert anode measured potential (vs reference) to RHE: V_anode_RHE = anode_measured_potential_vs_ref + ref_pot + 0.059 * anode_pH 3. Calculate cathode RHE (before IR correction): V_cathode_RHE = (V_anode_RHE + membrane_loss + Nern_pH_loss) - full_cell_V 4. Interpolate cathode resistance from calibration data (or use custom value) 5. Apply IR correction: V_cathode_RHE = V_cathode_RHE - (i/1000 * R * A) where i is current density in A/cm², R is interpolated resistance, A is geometric area Parameters: ----------- custom_anode_potential_vs_ref : float, optional Custom anode measured potential vs reference (V). If provided, overrides interpolation. custom_R_cathode : float, optional Custom cathode resistance (Ω). If provided, overrides interpolation. """ # Step 1: Interpolate anode measured potential vs reference (or use custom value) if custom_anode_potential_vs_ref is not None: anode_measured_potential_vs_ref = custom_anode_potential_vs_ref else: anode_measured_potential_vs_ref = interpolate_anode_potential_vs_ref(current_density) # Step 2: Convert anode measured potential to RHE v_anode_rhe = anode_measured_potential_vs_ref + ref_pot + 0.059 * anode_pH # Step 3: Calculate cathode RHE with membrane and Nernst pH losses (before IR correction) v_cathode_rhe = (v_anode_rhe + membrane_loss + Nern_pH_loss) - vcell # Step 4: Interpolate cathode resistance from calibration data (or use custom value) if custom_R_cathode is not None: R = custom_R_cathode else: R = interpolate_cathode_R(current_density) # current_density in mA/cm², R in ohm # Step 5: Apply IR correction # Convert current density from mA/cm² to A/cm² and apply IR correction # i/1000 converts mA/cm² to A/cm² IR_drop = (current_density / 1000.0) * R * geo_area v_cathode_rhe = v_cathode_rhe - IR_drop return v_cathode_rhe def get_overpotential(pot, pot_theory): overpot = abs(pot-pot_theory) return overpot def fullcell2halfcell(vcell, current_density, custom_params=None): ''' Main function to convert a voltage value from full cell to half cell vs she or rhe Parameters: ----------- vcell : float Full cell voltage (V) current_density : float Current density (mA/cm²) custom_params : dict, optional Custom parameters for voltage conversion ''' cali_dict = load_calibration_data_for_voltage_conversion(custom_params) # Extract custom values if provided custom_anode_pot = custom_params.get('anode_measured_potential_vs_ref') if custom_params else None custom_R = custom_params.get('R_cathode') if custom_params else None urhe = cell2rhe(vcell, cali_dict['conditions']['ref pot'], cali_dict['conditions']['anode pH'], cali_dict['conditions']['membrane loss'], cali_dict['conditions']['Nern pH loss'], current_density, cali_dict['conditions']['geo area'], custom_anode_potential_vs_ref=custom_anode_pot, custom_R_cathode=custom_R) # Use cathode_pH from calibration dict (which includes custom params if provided) ushe = rhe2she(urhe, cali_dict['conditions']['cathode pH'], cali_dict['conditions']['ref pot']) return ushe, urhe def convert_atomic_to_weight_fraction(df, element_columns): """ Convert atomic fraction to weight fraction for elemental compositions. """ df_converted = df.copy() for col in element_columns: if col in df_converted.columns and col in ATOMIC_WEIGHTS: df_converted[col] = df_converted[col] * ATOMIC_WEIGHTS[col] # Normalize to get weight fractions (0-1 scale) for idx, row in df_converted.iterrows(): total_weight = sum(row[col] for col in element_columns if col in df_converted.columns and col in ATOMIC_WEIGHTS) if total_weight > 0: for col in element_columns: if col in df_converted.columns and col in ATOMIC_WEIGHTS: df_converted.at[idx, col] = row[col] / total_weight return df_converted def load_original_data(): """Load the original data from CSV file or current data from dashboard""" try: # First try to load current data from dashboard current_data_file = "Data/current_data_her.json" if os.path.exists(current_data_file): with open(current_data_file, 'r') as f: saved_data = json.load(f) if isinstance(saved_data, dict) and 'data' in saved_data and 'columns' in saved_data: current_data = saved_data['data'] column_order = saved_data['columns'] df = pd.DataFrame(current_data, columns=column_order) elif isinstance(saved_data, list): df = pd.DataFrame(saved_data) else: df = pd.DataFrame(saved_data) # Filter for HER reaction if reaction column exists if 'reaction' in df.columns: df = df[df['reaction'] == 'HER'].copy() df = df.drop('reaction', axis=1) print(f"DEBUG: Available columns after loading HER data: {list(df.columns)}") print(f"DEBUG: Data shape: {df.shape}") print(f"DEBUG: Voltage columns present: {[col for col in df.columns if 'voltage' in col.lower()]}") return df except Exception as e: print(f"Could not load current data: {e}") # Fallback to original CSV data try: df = pd.read_csv("Data/DashboardData.csv") if 'reaction' in df.columns: df = df[df['reaction'] == 'HER'].copy() df = df.drop('reaction', axis=1) return df except Exception as e: print(f"Could not load CSV data: {e}") return pd.DataFrame() def calculate_pca_components(df): """Calculate PCA components from elemental composition data.""" if df.empty or len(df) < 2: df['PCA1'] = np.nan df['PCA2'] = np.nan return df # Get only elemental composition columns voltage_cols_to_exclude = ['voltage_mean', 'voltage_std', 'voltage'] composition_col = 'xrf composition' if 'xrf composition' in df.columns else 'target composition' element_cols = [col for col in df.columns if col not in ['sample id', 'source', 'batch number', 'batch date', 'current density', composition_col, 'target composition', 'xrf composition', 'rep'] + voltage_cols_to_exclude and not col.startswith('fe_') and not col.startswith('partial_current_') and not col.startswith('max_partial_current_') and not col.endswith('std')] # Filter out non-numeric columns numeric_element_cols = [] for col in element_cols: try: if pd.to_numeric(df[col], errors='coerce').notna().sum() >= 2: numeric_element_cols.append(col) except: continue if len(numeric_element_cols) < 2: df['PCA1'] = np.nan df['PCA2'] = np.nan return df # Prepare data for PCA pca_data = df[numeric_element_cols].copy() for col in pca_data.columns: pca_data[col] = pd.to_numeric(pca_data[col], errors='coerce') pca_data = pca_data.fillna(0) if pca_data.sum().sum() == 0: df['PCA1'] = 0 df['PCA2'] = 0 return df try: # Standardize and apply PCA scaler = StandardScaler() pca_data_scaled = scaler.fit_transform(pca_data) pca = PCA(n_components=2) pca_components = pca.fit_transform(pca_data_scaled) df['PCA1'] = pca_components[:, 0] df['PCA2'] = pca_components[:, 1] except Exception as e: print(f"PCA calculation failed: {e}") df['PCA1'] = np.nan df['PCA2'] = np.nan return df def format_column_name(column_name): """Format column names to be more readable.""" if column_name in ['voltage_mean', 'voltage']: return 'Full Cell Voltage (V)' elif column_name == 'voltage_she': return 'Est. Half-cell potential vs SHE (V)' elif column_name == 'voltage_rhe': return 'Est. Half-cell potential vs RHE (V)' elif column_name.startswith('fe_'): base_name = column_name.replace('fe_', '').replace('_mean', '') if base_name == 'h2': return 'Faradaic Efficiency H₂' elif base_name == 'co': return 'Faradaic Efficiency CO' elif base_name == 'ch4': return 'Faradaic Efficiency CH₄' elif base_name == 'c2h4': return 'Faradaic Efficiency C₂H₄' elif base_name == 'gas_total': return 'Faradaic Efficiency Gas Total' elif base_name == 'liquid': return 'Faradaic Efficiency Liquid' else: return 'Faradaic Efficiency ' + base_name.upper() elif column_name == 'cost_per_gram': return 'Cost per kg' elif column_name in ['PCA1', 'PCA2']: return column_name elif column_name in ['Ag', 'Au', 'Cd', 'Cu', 'Ga', 'Hg', 'In', 'Ni', 'Pd', 'Pt', 'Rh', 'Sn', 'Tl', 'Zn']: return column_name else: return column_name def find_pd_mean_value(df, target_column, source): """Find the mean value of a target column for a specific source where Pd composition is 1.0""" # Filter data for the specific source and Pd = 1.0 filtered_data = df[(df['source'] == source) & (df['Pd'] == 1.0)] if filtered_data.empty: return None # Get the mean value of the target column mean_value = filtered_data[target_column].mean() return mean_value def load_xrd_data(sample_id, data_type="raw"): """Load XRD data for a specific sample ID from Data/XRD or Data/CustomXRD directory.""" try: # First check for custom XRD data, then fall back to original custom_xrd_base = "Data/CustomXRD" original_xrd_base = "Data/XRD" # Construct potential file paths if data_type == "raw": custom_path = f"{custom_xrd_base}/raw/{sample_id}.xy" original_path = f"{original_xrd_base}/raw/{sample_id}.xy" elif data_type == "normalized": custom_path = f"{custom_xrd_base}/normalized/{sample_id}.csv" original_path = f"{original_xrd_base}/normalized/{sample_id}.csv" else: print(f"Invalid data type: {data_type}") return None # Check custom XRD first, then original xrd_file_path = None if os.path.exists(custom_path): xrd_file_path = custom_path print(f"DEBUG: Using custom XRD file: {custom_path}") elif os.path.exists(original_path): xrd_file_path = original_path print(f"DEBUG: Using original XRD file: {original_path}") else: print(f"XRD file not found in custom or original locations for sample {sample_id} ({data_type})") return None # Read the file data = [] with open(xrd_file_path, 'r') as f: lines = f.readlines() # Skip the first line (header) for line_num, line in enumerate(lines[1:], 2): # Start from line 2 line = line.strip() if line and not line.startswith('#'): # Skip empty lines and comments try: # Handle different separators (space, tab, comma) parts = line.replace(',', ' ').split() if len(parts) >= 2: x_val = float(parts[0]) y_val = float(parts[1]) data.append([x_val, y_val]) except ValueError: # Skip lines that can't be parsed as numbers if line_num <= 10: # Only log first few errors to avoid spam print(f"Warning: Could not parse line {line_num} in {xrd_file_path}: {line}") continue if not data: print(f"No valid data found in XRD file: {xrd_file_path}") return None print(f"Loaded XRD data for sample {sample_id} ({data_type}): {len(data)} data points") return data except Exception as e: print(f"Error loading XRD data: {e}") return None @her_plot_bp.route('/') def her_plot_main(): """Main HER plot page""" # Load and process data current_df = load_original_data() if current_df.empty: return "
Please ensure HER data is available in the main dashboard.
" # Calculate PCA components df_with_pca = calculate_pca_components(current_df) # Identify element columns voltage_cols_to_exclude = ['voltage_mean', 'voltage_std', 'voltage'] composition_col = 'xrf composition' if 'xrf composition' in df_with_pca.columns else 'target composition' element_cols = [col for col in df_with_pca.columns if col not in ['sample id', 'source', 'batch number', 'batch date', 'current density', composition_col, 'target composition', 'xrf composition', 'PCA1', 'PCA2', 'rep'] + voltage_cols_to_exclude and not col.startswith('fe_') and not col.startswith('partial_current_') and not col.startswith('max_partial_current_') and not col.endswith('std')] # Add PCA1 as first option if available if 'PCA1' in df_with_pca.columns: element_cols.insert(0, 'PCA1') if 'Cu' in df_with_pca.columns and 'Cu' not in element_cols: element_cols.insert(0, 'Cu') # Determine voltage column if 'voltage_mean' in df_with_pca.columns: y_axis_column = 'voltage_mean' elif 'voltage' in df_with_pca.columns: y_axis_column = 'voltage' else: return "Required voltage column not found.
" # Generate element options for dropdown element_options = ''.join([f'' for col in element_cols]) # Create the HTML template exactly matching the original html_template = f'''/Data/XRD/raw/ or /Data/CustomXRD/raw/ directoriessample_001.xy for raw or sample_001.csv for normalized)