""" AI-LOMAC platform-independent control-logic example =================================================== Code-availability note ---------------------- The complete experimental control software communicates with commercial battery cyclers and potentiostats through vendor-specific hardware APIs. Those interfaces, device drivers, authentication details, and instrument command sequences cannot be distributed because of commercial/licensing restrictions. This file therefore extracts the platform-independent analysis and decision logic used in the AI-LOMAC workflow for demonstration and reproducibility. All hardware interaction is intentionally omitted. The example starts from a cycle-resolved surface-capacitance trajectory, Cs(t), that has already been obtained from operando DEIS measurements. Instead of sending commands to an instrument, the code returns abstract control actions such as: - CONTINUE_SCHEDULED_FAST_CHARGE - SWITCH_TO_MITIGATION_CURRENT - STOP_AT_HARD_CUTOFF The provider-specific VLM network call is also represented by an optional callback. This keeps the example independent of service credentials and API syntax. In the experiments described in the manuscript, Gemini-3-pro-image-preview was used for VLM-assisted baseline localization. Workflow represented here ------------------------- 1. Receive the cycle-resolved Cs trajectory. 2. Smooth the trajectory and estimate a candidate pre-onset baseline using shape-adaptive piecewise quadratic-to-linear regression. 3. Use R^2 as the routing confidence: R^2 >= 0.97 -> accept the deterministic baseline. R^2 < 0.97 -> invoke VLM-assisted baseline localization once per cycle. 4. If the VLM request fails, times out, or returns no valid value, retain the locally estimated baseline boundary. 5. Calculate local Cs slopes using a five-point rolling least-squares window. 6. Define the warning threshold from the accepted baseline slopes: theta = Q3 + 1.5 * IQR 7. Confirm lithium-plating onset only when five consecutive slope estimates exceed theta. 8. If onset is confirmed, return a command to switch from the scheduled fast-charge current to the predefined mitigation current. Otherwise, continue the scheduled fast charge. 9. Independent hardware-level voltage cut-offs remain higher-priority safety constraints throughout charging and discharging; they are represented here by the `hard_cutoff_reached` flag rather than by vendor-specific API calls. Important scope note -------------------- This is an illustrative control-logic extraction, not the complete experimental software. DEIS acquisition, conversion from impedance to Cs, instrument timing, current/voltage programming, communication retries, and proprietary hardware APIs are intentionally not included. Expected offline input ---------------------- For the optional command-line demonstration, a CSV file should contain: CycleNum, TimeInStep_s, Capacitance_F The example processes each cycle independently and writes an analysis summary. """ from __future__ import annotations from dataclasses import dataclass from enum import Enum from pathlib import Path from typing import Callable, Dict, Optional, Tuple import numpy as np import pandas as pd # Matplotlib is needed only to prepare the trajectory image that would be passed # to the VLM. It is not required for the deterministic detector itself. import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt # ============================================================================= # Parameters # ============================================================================= @dataclass(frozen=True) class LogicConfig: """Configuration for the platform-independent AI-LOMAC logic.""" # Confidence-aware routing. r2_acceptance_threshold: float = 0.97 # The supplied implementation used a 1000 s deterministic fitting window. # This limit is protocol-specific and can be changed for a different test. deterministic_fit_end_s: float = 1000.0 # The manuscript specifies a 0-1800 s rendered trajectory for the 1C VLM # implementation shown in Supplementary Fig. 2. vlm_render_end_s: float = 1800.0 # Smoothing used before piecewise fitting in the supplied implementation. baseline_smoothing_window: int = 5 # Manuscript: n = 5 points for each rolling least-squares slope estimate. slope_window_points: int = 5 # Manuscript: theta = Q3 + 1.5 * IQR. iqr_factor: float = 1.5 # Manuscript: m = 5 consecutive suprathreshold slope estimates. consecutive_exceedances: int = 5 # Numerical guard for piecewise fitting. minimum_piecewise_points: int = 10 # Fixed prompt used for VLM-assisted localization. VLM_BASELINE_PROMPT = ( "Check whether the curve develops a nearly straight-line region toward the end. " "Identify the time at which the end segment begins to appear approximately linear. " "Ignore small local fluctuations, noise, or minor dips near the end, and base the " "judgment on the overall end-trend rather than subtle local changes. " "Return a single estimated start time for the near-linear region. ONLY return a number." ) class ControlAction(str, Enum): """Abstract actions returned by the platform-independent controller.""" WAIT_FOR_MORE_DATA = "WAIT_FOR_MORE_DATA" CONTINUE_SCHEDULED_FAST_CHARGE = "CONTINUE_SCHEDULED_FAST_CHARGE" SWITCH_TO_MITIGATION_CURRENT = "SWITCH_TO_MITIGATION_CURRENT" STOP_AT_HARD_CUTOFF = "STOP_AT_HARD_CUTOFF" @dataclass class BaselineResult: """Accepted baseline and routing information for one cycle.""" start_time_s: Optional[float] end_time_s: Optional[float] local_start_time_s: Optional[float] fit_r2: Optional[float] used_vlm: bool vlm_attempted: bool source: str @dataclass class DetectionResult: """Lithium-plating warning result for one cycle.""" warning: bool onset_time_s: Optional[float] threshold: Optional[float] max_consecutive_exceedances: int @dataclass class CycleResult: """Combined analysis and abstract control decision.""" cycle_num: int baseline: BaselineResult detection: DetectionResult action: ControlAction # A VLM provider receives an image path and the fixed prompt, and should return # the estimated start time of the near-linear terminal segment in seconds. # # Example signature: # def my_vlm_provider(image_path: str, prompt: str) -> Optional[float]: # ... # # The provider-specific HTTP request is intentionally not included here. VLMProvider = Callable[[str, str], Optional[float]] # ============================================================================= # Core numerical helpers # ============================================================================= def rolling_mean(values: np.ndarray, window: int) -> np.ndarray: """Causal rolling mean used before piecewise fitting.""" return ( pd.Series(values, dtype=float) .rolling(window=window, min_periods=1, center=False) .mean() .to_numpy() ) def rolling_least_squares_slope( times: np.ndarray, values: np.ndarray, window: int, ) -> np.ndarray: """ Calculate local slopes with an n-point rolling least-squares regression. This follows the manuscript description and is distinct from the separate requirement for five consecutive suprathreshold slope estimates. """ times = np.asarray(times, dtype=float) values = np.asarray(values, dtype=float) slopes = np.full(len(values), np.nan, dtype=float) if len(values) < 2: return slopes for i in range(2, min(window, len(values)) + 1): x = times[:i] y = values[:i] slopes[i - 1] = np.polyfit(x, y, 1)[0] for i in range(window, len(values) + 1): x = times[i - window:i] y = values[i - window:i] slopes[i - 1] = np.polyfit(x, y, 1)[0] if len(slopes) > 1 and np.isfinite(slopes[1]): slopes[0] = slopes[1] return slopes def q3_plus_iqr_threshold(values: np.ndarray, factor: float = 1.5) -> float: """Return Q3 + factor * IQR for finite values.""" values = np.asarray(values, dtype=float) values = values[np.isfinite(values)] if len(values) == 0: raise ValueError("No finite baseline slope values are available.") q1, q3 = np.percentile(values, [25.0, 75.0]) return float(q3 + factor * (q3 - q1)) def maximum_true_run(mask: np.ndarray) -> int: """Return the maximum number of consecutive True values.""" best = 0 current = 0 for flag in np.asarray(mask, dtype=bool): if flag: current += 1 best = max(best, current) else: current = 0 return best def first_confirmed_run_start( times: np.ndarray, mask: np.ndarray, required_length: int, ) -> Optional[float]: """ Return the first time at which a qualifying run starts. The reported onset is the first point of the first sequence containing at least `required_length` consecutive suprathreshold slope estimates. """ run = 0 run_start = None for i, flag in enumerate(np.asarray(mask, dtype=bool)): if flag: if run == 0: run_start = i run += 1 if run >= required_length: return float(times[run_start]) else: run = 0 run_start = None return None # ============================================================================= # Shape-adaptive deterministic baseline # ============================================================================= def fit_piecewise_quadratic_to_linear( times: np.ndarray, cs: np.ndarray, search_end_s: float, smoothing_window: int, minimum_points: int, ) -> Tuple[Optional[float], float]: """ Estimate the start of the near-linear pre-onset baseline. The trajectory is smoothed first. Candidate split points are evaluated using a quadratic fit before the split and a linear fit after the split. The split with the smallest total residual sum of squares is selected. Returns ------- baseline_start_s: Candidate start time of the near-linear terminal segment. r2: Goodness of fit of the combined piecewise model. """ times = np.asarray(times, dtype=float) cs = np.asarray(cs, dtype=float) mask = np.isfinite(times) & np.isfinite(cs) & (times <= search_end_s) x = times[mask] y = cs[mask] if len(x) < minimum_points: return None, 0.0 order = np.argsort(x) x = x[order] y = rolling_mean(y[order], smoothing_window) sst = np.sum((y - np.mean(y)) ** 2) if not np.isfinite(sst) or sst <= 0: return None, 0.0 best_sse = np.inf best_split = None # Keep at least five points on both sides, following the supplied # implementation. for split in range(5, len(x) - 5): x1, y1 = x[:split], y[:split] x2, y2 = x[split:], y[split:] try: quadratic = np.polyfit(x1, y1, 2) linear = np.polyfit(x2, y2, 1) except (np.linalg.LinAlgError, ValueError): continue pred1 = np.polyval(quadratic, x1) pred2 = np.polyval(linear, x2) sse = np.sum((y1 - pred1) ** 2) + np.sum((y2 - pred2) ** 2) if sse < best_sse: best_sse = sse best_split = split if best_split is None: return None, 0.0 r2 = 1.0 - best_sse / sst return float(x[best_split]), float(r2) # ============================================================================= # VLM fallback interface # ============================================================================= def render_vlm_input( times: np.ndarray, cs: np.ndarray, cycle_num: int, output_path: str, render_end_s: float, ) -> str: """ Render the Cs trajectory used for VLM-assisted baseline localization. The manuscript specifies a 0-1800 s rendering window for the 1C implementation shown in Supplementary Fig. 2. """ times = np.asarray(times, dtype=float) cs = np.asarray(cs, dtype=float) mask = np.isfinite(times) & np.isfinite(cs) & (times >= 0) & (times <= render_end_s) x = times[mask] y = cs[mask] if len(x) == 0: raise ValueError("No data are available inside the VLM rendering window.") fig, ax = plt.subplots(figsize=(8, 6)) ax.plot(x, y, linewidth=2) ax.set_title(f"Cycle {cycle_num}: Cs vs time (0-{render_end_s:.0f} s)") ax.set_xlabel("Time (s)") ax.set_ylabel("Capacitance, Cs (F)") ax.grid(True, linestyle="--", alpha=0.5) fig.tight_layout() fig.savefig(output_path, dpi=150) plt.close(fig) return output_path def resolve_baseline( times: np.ndarray, cs: np.ndarray, cycle_num: int, config: LogicConfig, vlm_provider: Optional[VLMProvider] = None, image_directory: Optional[str] = None, ) -> BaselineResult: """ Resolve one accepted baseline using confidence-aware routing. If R^2 >= 0.97, the deterministic result is accepted directly. If R^2 < 0.97, the VLM fallback is attempted once for this cycle. If the VLM does not return a valid value, the locally estimated boundary is retained, matching the fail-safe behavior described in the manuscript. """ times = np.asarray(times, dtype=float) cs = np.asarray(cs, dtype=float) if len(times) == 0 or np.nanmax(times) < config.deterministic_fit_end_s: return BaselineResult( start_time_s=None, end_time_s=None, local_start_time_s=None, fit_r2=None, used_vlm=False, vlm_attempted=False, source="insufficient_data", ) local_start, r2 = fit_piecewise_quadratic_to_linear( times=times, cs=cs, search_end_s=config.deterministic_fit_end_s, smoothing_window=config.baseline_smoothing_window, minimum_points=config.minimum_piecewise_points, ) if local_start is None: return BaselineResult( start_time_s=None, end_time_s=None, local_start_time_s=None, fit_r2=r2, used_vlm=False, vlm_attempted=False, source="no_valid_local_baseline", ) if r2 >= config.r2_acceptance_threshold: return BaselineResult( start_time_s=local_start, end_time_s=config.deterministic_fit_end_s, local_start_time_s=local_start, fit_r2=r2, used_vlm=False, vlm_attempted=False, source="deterministic", ) # Low-confidence route: VLM is invoked only once per cycle. vlm_attempted = True # If the 0-1800 s context is not yet available, retain the local boundary # temporarily rather than inventing a new value. if np.nanmax(times) < config.vlm_render_end_s: return BaselineResult( start_time_s=local_start, end_time_s=config.deterministic_fit_end_s, local_start_time_s=local_start, fit_r2=r2, used_vlm=False, vlm_attempted=False, source="local_pending_vlm_context", ) if vlm_provider is None: return BaselineResult( start_time_s=local_start, end_time_s=config.deterministic_fit_end_s, local_start_time_s=local_start, fit_r2=r2, used_vlm=False, vlm_attempted=vlm_attempted, source="local_fallback_no_vlm_provider", ) image_directory = image_directory or "." Path(image_directory).mkdir(parents=True, exist_ok=True) image_path = str(Path(image_directory) / f"cycle_{cycle_num}_vlm_input.png") try: render_vlm_input( times=times, cs=cs, cycle_num=cycle_num, output_path=image_path, render_end_s=config.vlm_render_end_s, ) vlm_start = vlm_provider(image_path, VLM_BASELINE_PROMPT) if vlm_start is not None and np.isfinite(vlm_start) and vlm_start >= 0: return BaselineResult( start_time_s=float(vlm_start), end_time_s=config.vlm_render_end_s, local_start_time_s=local_start, fit_r2=r2, used_vlm=True, vlm_attempted=vlm_attempted, source="vlm", ) except Exception: # Provider/network/timeout errors are intentionally collapsed to the # manuscript-specified fail-safe behavior: retain the local boundary. pass return BaselineResult( start_time_s=local_start, end_time_s=config.deterministic_fit_end_s, local_start_time_s=local_start, fit_r2=r2, used_vlm=False, vlm_attempted=vlm_attempted, source="local_fallback_after_vlm_failure", ) # ============================================================================= # Common deterministic knee-point detector # ============================================================================= def detect_knee_point( times: np.ndarray, cs: np.ndarray, baseline: BaselineResult, config: LogicConfig, ) -> DetectionResult: """ Detect the Cs knee point from the accepted baseline. Both deterministic and VLM-assisted baseline routes enter this same deterministic slope-thresholding and consecutive-confirmation procedure. """ if baseline.start_time_s is None or baseline.end_time_s is None: return DetectionResult( warning=False, onset_time_s=None, threshold=None, max_consecutive_exceedances=0, ) times = np.asarray(times, dtype=float) cs = np.asarray(cs, dtype=float) valid = np.isfinite(times) & np.isfinite(cs) times = times[valid] cs = cs[valid] if len(times) < config.slope_window_points: return DetectionResult(False, None, None, 0) order = np.argsort(times) times = times[order] cs = cs[order] slopes = rolling_least_squares_slope( times, cs, window=config.slope_window_points, ) baseline_mask = ( (times >= baseline.start_time_s) & (times <= baseline.end_time_s) & np.isfinite(slopes) ) baseline_slopes = slopes[baseline_mask] if len(baseline_slopes) == 0: return DetectionResult(False, None, None, 0) threshold = q3_plus_iqr_threshold( baseline_slopes, factor=config.iqr_factor, ) # The baseline defines the pre-onset reference distribution. Monitoring # continues after the baseline region using the same local-slope estimator. monitor_mask = (times > baseline.end_time_s) & np.isfinite(slopes) monitor_times = times[monitor_mask] monitor_slopes = slopes[monitor_mask] if len(monitor_slopes) == 0: return DetectionResult(False, None, threshold, 0) exceed = monitor_slopes > threshold max_run = maximum_true_run(exceed) onset_time = first_confirmed_run_start( monitor_times, exceed, required_length=config.consecutive_exceedances, ) return DetectionResult( warning=onset_time is not None, onset_time_s=onset_time, threshold=threshold, max_consecutive_exceedances=max_run, ) # ============================================================================= # Abstract control decision # ============================================================================= def decide_control_action( detection: DetectionResult, hard_cutoff_reached: bool = False, ) -> ControlAction: """ Convert the analytical result into an abstract control action. Hardware-level voltage cut-offs have priority. The actual current-setting command is intentionally omitted from this public example. """ if hard_cutoff_reached: return ControlAction.STOP_AT_HARD_CUTOFF if detection.warning: return ControlAction.SWITCH_TO_MITIGATION_CURRENT return ControlAction.CONTINUE_SCHEDULED_FAST_CHARGE def analyze_cycle( cycle_num: int, times: np.ndarray, cs: np.ndarray, config: LogicConfig = LogicConfig(), vlm_provider: Optional[VLMProvider] = None, image_directory: Optional[str] = None, hard_cutoff_reached: bool = False, ) -> CycleResult: """ Run the complete platform-independent analysis-to-control logic for one cycle. """ baseline = resolve_baseline( times=times, cs=cs, cycle_num=cycle_num, config=config, vlm_provider=vlm_provider, image_directory=image_directory, ) if baseline.start_time_s is None: detection = DetectionResult(False, None, None, 0) action = ( ControlAction.STOP_AT_HARD_CUTOFF if hard_cutoff_reached else ControlAction.WAIT_FOR_MORE_DATA ) return CycleResult(cycle_num, baseline, detection, action) detection = detect_knee_point( times=times, cs=cs, baseline=baseline, config=config, ) action = decide_control_action( detection=detection, hard_cutoff_reached=hard_cutoff_reached, ) return CycleResult( cycle_num=cycle_num, baseline=baseline, detection=detection, action=action, ) # ============================================================================= # Optional offline demonstration # ============================================================================= def load_cycle_csv(csv_path: str) -> pd.DataFrame: """Load the simplified offline demonstration format.""" required = ["CycleNum", "TimeInStep_s", "Capacitance_F"] df = pd.read_csv(csv_path) df.columns = [str(c).strip() for c in df.columns] missing = [c for c in required if c not in df.columns] if missing: raise ValueError( "The CSV file is missing required columns: " + ", ".join(missing) ) df = df[required].copy() for col in required: df[col] = pd.to_numeric(df[col], errors="coerce") df = df.dropna(subset=required) if df.empty: raise ValueError("No valid numerical rows were found in the CSV file.") return df def process_offline_csv( csv_path: str, output_csv_path: str, config: LogicConfig = LogicConfig(), ) -> pd.DataFrame: """ Demonstrate the extracted logic on pre-recorded cycle-resolved Cs data. No hardware commands and no external VLM calls are made in this default demonstration. Low-confidence trajectories therefore follow the documented fail-safe path and retain the local baseline estimate. """ df = load_cycle_csv(csv_path) rows = [] image_dir = str(Path(output_csv_path).with_suffix("")) + "_vlm_inputs" for cycle_num, group in df.groupby("CycleNum", sort=True): group = ( group.sort_values("TimeInStep_s") .drop_duplicates(subset="TimeInStep_s", keep="first") ) times = group["TimeInStep_s"].to_numpy(dtype=float) cs = group["Capacitance_F"].to_numpy(dtype=float) result = analyze_cycle( cycle_num=int(cycle_num), times=times, cs=cs, config=config, vlm_provider=None, # Provider-specific API intentionally omitted. image_directory=image_dir, hard_cutoff_reached=False, ) rows.append( { "CycleNum": result.cycle_num, "Baseline_Start_s": result.baseline.start_time_s, "Baseline_End_s": result.baseline.end_time_s, "Fit_R2": result.baseline.fit_r2, "Used_VLM": result.baseline.used_vlm, "Baseline_Source": result.baseline.source, "Warning": result.detection.warning, "Onset_Time_s": result.detection.onset_time_s, "Slope_Threshold": result.detection.threshold, "Max_Consecutive_Exceedances": ( result.detection.max_consecutive_exceedances ), "Control_Action": result.action.value, } ) out = pd.DataFrame(rows) out.to_csv(output_csv_path, index=False) return out if __name__ == "__main__": import argparse parser = argparse.ArgumentParser( description=( "Offline demonstration of the platform-independent AI-LOMAC " "analysis and control logic. Hardware APIs are intentionally omitted." ) ) parser.add_argument( "input_csv", help="CSV containing CycleNum, TimeInStep_s, and Capacitance_F.", ) parser.add_argument( "--output", default="AI_LOMAC_logic_results.csv", help="Output CSV path.", ) args = parser.parse_args() results = process_offline_csv( csv_path=args.input_csv, output_csv_path=args.output, ) print(results.to_string(index=False)) print(f"\nResults written to: {args.output}")