Battery Data Analysis

name: battery-data-analysis description: 'Analyze robot battery performance data. Use when plotting discharge curves, current histograms, or estimating state of health.'

Battery Performance Analysis for Robots

Collecting Battery Data

# Record battery state to a rosbag
ros2 bag record --storage mcap /battery_state /cmd_vel /odom

# BatteryState message fields (sensor_msgs/msg/BatteryState):
#   float32 voltage          — pack voltage (V)
#   float32 current          — current draw, negative = discharging (A)
#   float32 charge           — remaining charge (Ah)
#   float32 capacity         — full capacity (Ah)
#   float32 percentage       — 0.0 to 1.0
#   float32 temperature      — battery temp (°C)
#   uint8   power_supply_status  — CHARGING/DISCHARGING/NOT_CHARGING/FULL
#   float32[] cell_voltage   — per-cell voltages if available

Key Metrics

Voltage Thresholds by Chemistry

Analysis Script

#!/usr/bin/env python3
"""Battery performance analysis: discharge curve, current profile, power vs velocity."""

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from rclpy.serialization import deserialize_message
from rosidl_runtime_py.utilities import get_message
import rosbag2_py


def read_topic(bag_path: str, topic: str):
    """Yield (msg, timestamp_ns) for a single topic."""
    reader = rosbag2_py.SequentialReader()
    reader.open(
        rosbag2_py.StorageOptions(uri=bag_path, storage_id=""),
        rosbag2_py.ConverterOptions("cdr", "cdr"),
    )
    type_map = {t.name: t.type for t in reader.get_all_topics_and_types()}
    reader.set_filter(rosbag2_py.StorageFilter(topics=[topic]))
    msg_type = get_message(type_map[topic])
    while reader.has_next():
        _, data, ts = reader.read_next()
        yield deserialize_message(data, msg_type), ts


def load_battery_data(bag_path: str) -> pd.DataFrame:
    rows = []
    for msg, ts in read_topic(bag_path, "/battery_state"):
        rows.append({
            "time_ns": ts,
            "voltage": msg.voltage,
            "current": abs(msg.current),  # make positive for discharge
            "percentage": msg.percentage * 100,
            "temperature": msg.temperature if msg.temperature > -50 else np.nan,
            "charge": msg.charge,
            "capacity": msg.capacity,
        })
    df = pd.DataFrame(rows)
    df["time_s"] = (df["time_ns"] - df["time_ns"].iloc[0]) / 1e9
    df["time_min"] = df["time_s"] / 60
    df["power_w"] = df["voltage"] * df["current"]
    return df


def load_velocity_data(bag_path: str) -> pd.DataFrame:
    rows = []
    for msg, ts in read_topic(bag_path, "/cmd_vel"):
        rows.append({
            "time_ns": ts,
            "vx": msg.linear.x,
            "wz": msg.angular.z,
        })
    df = pd.DataFrame(rows)
    df["time_s"] = (df["time_ns"] - df["time_ns"].iloc[0]) / 1e9
    df["speed"] = np.sqrt(df["vx"]**2 + (df["wz"] * 0.2)**2)  # approximate
    return df


def analyze_battery(bag_path: str):
    bat = load_battery_data(bag_path)

    # Compute statistics
    total_energy_wh = np.trapz(bat["power_w"], bat["time_s"]) / 3600
    mean_current = bat["current"].mean()
    peak_current = bat["current"].max()
    p95_current = bat["current"].quantile(0.95)
    voltage_drop = bat["voltage"].iloc[0] - bat["voltage"].iloc[-1]
    duration_min = bat["time_min"].iloc[-1]

    print(f"Duration:        {duration_min:.1f} min")
    print(f"Energy consumed: {total_energy_wh:.2f} Wh")
    print(f"Mean current:    {mean_current:.2f} A")
    print(f"Peak current:    {peak_current:.2f} A (P95: {p95_current:.2f} A)")
    print(f"Voltage drop:    {voltage_drop:.2f} V ({bat['voltage'].iloc[0]:.1f}V → {bat['voltage'].iloc[-1]:.1f}V)")

    if bat["capacity"].iloc[0] > 0:
        soh = bat["capacity"].iloc[0] / bat["capacity"].max() * 100
        print(f"State of Health: {soh:.1f}%")

    fig, axes = plt.subplots(2, 2, figsize=(14, 10))

    # 1. Discharge curve
    ax = axes[0, 0]
    ax.plot(bat["time_min"], bat["voltage"], "b-", linewidth=1)
    ax.axhline(y=10.0, color="r", linestyle="--", alpha=0.7, label="Min cutoff (LiFePO4)")
    ax.set_xlabel("Time (min)")
    ax.set_ylabel("Voltage (V)")
    ax.set_title("Discharge Curve")
    ax.legend()
    ax.grid(True, alpha=0.3)

    # 2. Current histogram
    ax = axes[0, 1]
    ax.hist(bat["current"], bins=80, edgecolor="black", alpha=0.7, color="steelblue")
    ax.axvline(mean_current, color="r", linestyle="--", label=f"Mean: {mean_current:.2f}A")
    ax.axvline(p95_current, color="orange", linestyle="--", label=f"P95: {p95_current:.2f}A")
    ax.set_xlabel("Current (A)")
    ax.set_ylabel("Count")
    ax.set_title("Current Draw Distribution")
    ax.legend()
    ax.grid(True, alpha=0.3)

    # 3. Power and voltage over time (voltage sag visible)
    ax = axes[1, 0]
    ax.plot(bat["time_min"], bat["power_w"], "g-", linewidth=0.5, alpha=0.7, label="Power (W)")
    ax2 = ax.twinx()
    ax2.plot(bat["time_min"], bat["voltage"], "b-", linewidth=0.8, alpha=0.5, label="Voltage (V)")
    ax.set_xlabel("Time (min)")
    ax.set_ylabel("Power (W)", color="g")
    ax2.set_ylabel("Voltage (V)", color="b")
    ax.set_title("Power Draw & Voltage Sag")
    ax.grid(True, alpha=0.3)

    # 4. Power vs velocity correlation
    try:
        vel = load_velocity_data(bag_path)
        # Resample velocity to battery timestamps
        bat_resampled = bat.set_index("time_s")
        vel_resampled = vel.set_index("time_s").reindex(
            bat_resampled.index, method="nearest", tolerance=1.0
        )
        mask = vel_resampled["speed"].notna()
        ax = axes[1, 1]
        ax.scatter(vel_resampled.loc[mask, "speed"], bat_resampled.loc[mask, "power_w"],
                   alpha=0.3, s=5)
        ax.set_xlabel("Robot Speed (m/s approx)")
        ax.set_ylabel("Power (W)")
        ax.set_title("Power vs Velocity Correlation")
        ax.grid(True, alpha=0.3)
    except Exception:
        axes[1, 1].text(0.5, 0.5, "No /cmd_vel data", transform=axes[1, 1].transAxes,
                        ha="center", va="center", fontsize=14)

    plt.suptitle(f"Battery Analysis — {total_energy_wh:.1f} Wh over {duration_min:.0f} min", fontsize=14)
    plt.tight_layout()
    plt.savefig("battery_analysis.png", dpi=150)
    plt.show()


if __name__ == "__main__":
    import sys
    bag_path = sys.argv[1] if len(sys.argv) > 1 else "rosbag2_recording"
    analyze_battery(bag_path)

Voltage Sag Analysis

Voltage sag reveals internal resistance. Under load, voltage drops; at rest, it recovers:

# Detect sag events: voltage drops > 0.3V within 2 seconds
voltage_diff = bat["voltage"].diff()
sag_events = bat[voltage_diff < -0.3]
for _, event in sag_events.iterrows():
    print(f"Sag at t={event['time_min']:.1f}min: ΔV={voltage_diff.loc[event.name]:.2f}V, "
          f"I={event['current']:.2f}A → R_int≈{abs(voltage_diff.loc[event.name])/event['current']:.3f}Ω")

Automated Alerts

# Runtime alert thresholds
ALERTS = {
    "low_voltage": {"threshold": 10.5, "field": "voltage", "op": "lt"},
    "high_current": {"threshold": 15.0, "field": "current", "op": "gt"},
    "high_temp": {"threshold": 45.0, "field": "temperature", "op": "gt"},
    "low_soh": {"threshold": 70.0, "field": "capacity", "op": "lt_pct"},  # % of nominal
}

Tracking Degradation Over Charge Cycles

Log each full discharge cycle's total energy and capacity. Plot capacity vs cycle number to predict replacement schedule. A LiFePO4 pack typically retains 80% capacity after 2000+ cycles; LiPo after ~300-500 cycles.