File size: 10,297 Bytes
8b37c3f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 | """Adapters that normalize NASA, CALCE, and Oxford discharge cycles.
Every adapter yields the same :class:`DischargeCycle` record. Dataset-specific
parsing is isolated here so feature extraction and validation remain identical
across domains.
"""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable, Iterator
import numpy as np
import pandas as pd
@dataclass
class DischargeCycle:
dataset: str
battery_id: str
cycle_number: int
capacity_ah: float
ambient_temperature_c: float
measurements: pd.DataFrame
source: str
def _as_array(value: object) -> np.ndarray:
return np.asarray(value, dtype=float).reshape(-1)
def _matlab_datenum_to_elapsed_seconds(values: object) -> np.ndarray:
"""Convert MATLAB serial-day timestamps to elapsed seconds."""
days = _as_array(values)
return (days - days[0]) * 86_400.0
def _sanitize_temperature_channel(
values: object,
*,
ambient_temperature_c: float,
max_median_deviation_c: float = 20.0,
) -> np.ndarray:
"""Mark a grossly inconsistent whole-cycle temperature channel missing."""
temperature = _as_array(values)
finite = temperature[np.isfinite(temperature)]
if finite.size == 0 or abs(float(np.median(finite)) - ambient_temperature_c) > max_median_deviation_c:
return np.full(temperature.shape, np.nan)
return temperature
def iter_nasa_mat_cycles(paths: Iterable[str | Path]) -> Iterator[DischargeCycle]:
"""Yield discharge cycles from original NASA PCoE MATLAB files."""
from scipy.io import loadmat
next_cycle_number: dict[str, int] = {}
for raw_path in paths:
path = Path(raw_path)
battery_id = path.stem
payload = loadmat(path, squeeze_me=True, struct_as_record=False)
root = payload.get(battery_id)
if root is None:
candidates = [value for key, value in payload.items() if not key.startswith("__")]
if len(candidates) != 1:
raise ValueError(f"Could not identify battery structure in {path}")
root = candidates[0]
cycles = np.atleast_1d(root.cycle)
discharge_index = next_cycle_number.get(battery_id, 0)
for cycle in cycles:
cycle_type = str(cycle.type).strip().lower()
if cycle_type != "discharge":
continue
data = cycle.data
capacity_values = np.asarray(data.Capacity, dtype=float).reshape(-1)
if capacity_values.size != 1 or not np.isfinite(capacity_values[0]):
continue
capacity = float(capacity_values[0])
time = _as_array(data.Time)
voltage = _as_array(data.Voltage_measured)
current = _as_array(data.Current_measured)
temperature = _as_array(data.Temperature_measured)
lengths = [len(time), len(voltage), len(current)]
n = min(lengths)
if n < 3:
continue
if len(temperature) < n:
temperature = np.full(n, np.nan)
frame = pd.DataFrame({
"Time": time[:n],
"Voltage_measured": voltage[:n],
"Current_measured": current[:n],
"Temperature_measured": temperature[:n],
})
yield DischargeCycle(
dataset="NASA",
battery_id=battery_id,
cycle_number=discharge_index,
capacity_ah=capacity,
ambient_temperature_c=float(np.asarray(cycle.ambient_temperature).squeeze()),
measurements=frame,
source=str(path),
)
discharge_index += 1
next_cycle_number[battery_id] = discharge_index
def _calce_cycles_from_frame(
frame: pd.DataFrame,
*,
battery_id: str,
source: str,
ambient_temperature_c: float = 25.0,
) -> list[tuple[pd.Timestamp, DischargeCycle]]:
"""Extract candidate CALCE discharges from one Arbin worksheet."""
required = {"Test_Time(s)", "Date_Time", "Cycle_Index", "Current(A)", "Voltage(V)", "Discharge_Capacity(Ah)"}
missing = required.difference(frame.columns)
if missing:
raise KeyError(f"Missing CALCE columns: {sorted(missing)}")
records: list[tuple[pd.Timestamp, DischargeCycle]] = []
for raw_cycle, group in frame.groupby("Cycle_Index", sort=True):
discharge = group[pd.to_numeric(group["Current(A)"], errors="coerce") < -0.05].copy()
if len(discharge) < 3:
continue
all_capacity = pd.to_numeric(group["Discharge_Capacity(Ah)"], errors="coerce")
# Some Arbin exports continue the counter across steps/files. The
# within-cycle range is invariant to that offset and is the delivered Ah.
capacity = all_capacity.max() - all_capacity.min()
if not np.isfinite(capacity) or not 0.3 <= capacity <= 1.5:
continue
test_time = pd.to_numeric(discharge["Test_Time(s)"], errors="coerce")
measurements = pd.DataFrame({
"Time": test_time - test_time.iloc[0],
"Voltage_measured": pd.to_numeric(discharge["Voltage(V)"], errors="coerce"),
"Current_measured": pd.to_numeric(discharge["Current(A)"], errors="coerce"),
"Temperature_measured": np.nan,
}).dropna(subset=["Time", "Voltage_measured", "Current_measured"])
started = pd.to_datetime(discharge["Date_Time"], errors="coerce").min()
if pd.isna(started):
started = pd.Timestamp.min
records.append((started, DischargeCycle(
dataset="CALCE",
battery_id=battery_id,
cycle_number=int(raw_cycle),
capacity_ah=float(capacity),
ambient_temperature_c=float(ambient_temperature_c),
measurements=measurements.reset_index(drop=True),
source=source,
)))
return records
def iter_calce_xlsx_cycles(
root: str | Path,
*,
battery_ids: tuple[str, ...] = ("CS2_35", "CS2_36", "CS2_37", "CS2_38"),
ambient_temperature_c: float = 25.0,
) -> Iterator[DischargeCycle]:
"""Yield chronological, de-duplicated discharges from CALCE ZIP extracts."""
root = Path(root)
for battery_id in battery_ids:
candidates: list[tuple[pd.Timestamp, DischargeCycle]] = []
for path in sorted(root.glob(f"{battery_id}/**/*.xlsx")):
book = pd.ExcelFile(path, engine="openpyxl")
data_sheets = [name for name in book.sheet_names if name.lower() != "info"]
for sheet in data_sheets:
frame = pd.read_excel(book, sheet_name=sheet)
candidates.extend(_calce_cycles_from_frame(
frame,
battery_id=battery_id,
source=f"{path}:{sheet}",
ambient_temperature_c=ambient_temperature_c,
))
candidates.sort(key=lambda pair: pair[0])
seen: set[tuple[int, int]] = set()
ordered: list[DischargeCycle] = []
for started, record in candidates:
key = (int(started.value // 10**9), int(round(record.capacity_ah * 100_000)))
if key not in seen:
seen.add(key)
ordered.append(record)
for cycle_number, record in enumerate(ordered):
record.cycle_number = cycle_number
yield record
def iter_oxford_mat_cycles(path: str | Path) -> Iterator[DischargeCycle]:
"""Yield Oxford 1C characterization discharges from Cells 1-8."""
from scipy.io import loadmat
path = Path(path)
for cell_number in range(1, 9):
battery_id = f"Cell{cell_number}"
payload = loadmat(
path,
variable_names=[battery_id],
squeeze_me=True,
struct_as_record=False,
)
cell = payload[battery_id]
for cycle_name in sorted(cell._fieldnames, key=lambda name: int(name[3:])):
diagnostic = getattr(cell, cycle_name)
if not hasattr(diagnostic, "C1dc"):
continue
discharge = diagnostic.C1dc
time_s = _matlab_datenum_to_elapsed_seconds(discharge.t)
voltage = _as_array(discharge.v)
charge_mah = _as_array(discharge.q)
temperature = _sanitize_temperature_channel(
discharge.T,
ambient_temperature_c=40.0,
)
n = min(len(time_s), len(voltage), len(charge_mah), len(temperature))
if n < 3:
continue
time_s, voltage = time_s[:n], voltage[:n]
charge_mah, temperature = charge_mah[:n], temperature[:n]
capacity_ah = float(np.nanmax(charge_mah) - np.nanmin(charge_mah)) / 1000.0
with np.errstate(divide="ignore", invalid="ignore"):
current_a = np.gradient(charge_mah, time_s) * 3.6
current_a = (
pd.Series(current_a)
.replace([np.inf, -np.inf], np.nan)
.interpolate(limit_direction="both")
.to_numpy()
)
frame = pd.DataFrame({
"Time": time_s,
"Voltage_measured": voltage,
"Current_measured": current_a,
"Temperature_measured": temperature,
})
yield DischargeCycle(
dataset="Oxford",
battery_id=battery_id,
cycle_number=int(cycle_name[3:]),
capacity_ah=capacity_ah,
ambient_temperature_c=40.0,
measurements=frame,
source=f"{path}:{battery_id}/{cycle_name}/C1dc",
)
def cycle_inventory(cycles: Iterable[DischargeCycle]) -> pd.DataFrame:
"""Create the one-row-per-cycle audit table used by Notebook 01."""
return pd.DataFrame([
{
"dataset": cycle.dataset,
"battery_id": cycle.battery_id,
"cycle_number": cycle.cycle_number,
"capacity_ah": cycle.capacity_ah,
"ambient_temperature_c": cycle.ambient_temperature_c,
"n_measurements": len(cycle.measurements),
"source": cycle.source,
}
for cycle in cycles
])
|