jellyfish-powersupply/sw/jf_battery_profiler.py

#!/usr/bin/env python

# SPDX-FileCopyrightText: 2025 Igor Brkic <igor@hyperglitch.com>
#
# SPDX-License-Identifier: GPL-3.0-or-later

import json
import os
import queue
import select
import serial
import sys
import time

from libjellyfishopp import JfUsbProcess

DECIMATION_FACTOR = 10

DISCHARGE_CURRENT_MA = 200
REST_CURRENT_MA = 20
DISCHARGE_DURATION_S = 5
REST_DURATION_S = 5

vout_key = 'adc10'
iout_key = 'adc00'


# First step before profiling is to determine the output voltage at which the desired
# currents are reached. This is done by scanning the output voltage from the battery's
# open circuit voltage down towards the 0V.
# For small coincell batteries, each current pulse requires a rest time before being
# able to provide the next pulse.
VOUT_SCAN_DECR_MV = 100 # how much to decrement the output voltage when scanning
                        # for the discharge and rest current thresholds

VOUT_SCAN_DURATION_S = 0.2 # how long to keep the discharge while scanning
VOUT_SCAN_REST_S = 4 # how long to wait between checks

def timed_input(prompt='', timeout=1.0):
    #print(prompt, end='', flush=True)
    rlist, _, _ = select.select([sys.stdin], [], [], timeout)
    if rlist:
        return sys.stdin.readline().strip()
    return None

def queue_flush(q):
    while True:
        try:
            q.get_nowait()
        except queue.Empty:
            break


def get_averaged_data(data_queue, timeout=0.1, cycles=1):
    count = 0
    out = {
            'adc00': 0.0,
            'adc01': 0.0,
            'adc10': 0.0,
            'adc11': 0.0,
            }
    for _ in range(cycles):
        try:
            q = data_queue.get(timeout=timeout)
        except queue.Empty:
            print("ERROR: No data in the queue")
            continue
        for rec in q:
            out['adc00'] += float(rec['adc00'])
            out['adc01'] += float(rec['adc01'])
            out['adc10'] += float(rec['adc10'])
            out['adc11'] += float(rec['adc11'])
            count += 1

    if count > 0:
        out['adc00'] /= count
        out['adc01'] /= count
        out['adc10'] /= count
        out['adc11'] /= count
    return out


def scan_thresholds(scpi, data_queue, vout):
    vout_discharge = -1
    vout_rest = -1
    count = 0
    vout_set = vout
    while True:
        # set and enable the output
        print(f"Setting output voltage to {int(vout_set)}mV")
        scpi.write(b"SOURCE:VOLT %d\n"%(int(vout_set),))
        time.sleep(0.1)
        scpi.write(b"ROUTE:OUT 1\n")
        time.sleep(VOUT_SCAN_DURATION_S)

        # read measurements
        queue_flush(data_queue)
        time.sleep(0.1)
        rec = get_averaged_data(data_queue)
        iout = rec[iout_key]
        vout = rec[vout_key]
        print(f"\rCurrent: {iout:.3f}mA, Output voltage: {vout/1000.0:.3f}V")
        queue_flush(data_queue)
        count += 1

        if vout_rest == -1 and abs(iout) > REST_CURRENT_MA:
            vout_rest = vout_set+VOUT_SCAN_DECR_MV//2  # current value is too high so
                                                                # we need to scan a bit more
            print("   == Found rest current threshold: %.3fV"%(vout_rest/1000.0))
        if vout_discharge == -1 and abs(iout) > DISCHARGE_CURRENT_MA:
            vout_discharge = vout_set+VOUT_SCAN_DECR_MV//2 # current value is too high
                                                                    # so we need to scan a bit more
            print("   == Found discharge current threshold: %.3fV"%(vout_discharge/1000.0))
            break

        # disable the output
        scpi.write(b"ROUTE:OUT 0\n")
        print(f"Letting the battery to rest for {VOUT_SCAN_REST_S}s")
        time.sleep(VOUT_SCAN_REST_S) # give it time to rest
        vout_set -= VOUT_SCAN_DECR_MV # add some heuristic here to adjust the decrement better

        if vout_set < 200:
            break

    return vout_discharge, vout_rest


def main():
    scpi_interface = '/dev/ttyUSB0'
    scpi_baud = 230400

    calibration_table_full = {}
    if os.path.exists('calibration.json'):
        print("Loading calibration table...")
        with open('calibration.json') as f:
            calibration_table_full = json.load(f)

    else:
        print("No calibration table found, plotting raw values")

    try:
        scpi = serial.Serial(scpi_interface, scpi_baud, timeout=0.1)
    except serial.SerialException:
        print("ERROR: Could not open serial port")
        return

    scpi.write(b"SYSTEM:ID?\n")
    try:
        devid = scpi.readlines()[-1].strip().decode()
    except (TypeError, IndexError, AttributeError):
        print("ERROR: Could not read device ID")
        return

    print(f"Device ID: {devid}")

    try:
        calibration_table = calibration_table_full[devid]
    except KeyError:
        print("ERROR: Could not find calibration table for this device")
        return

    proc = JfUsbProcess(calibration_table=calibration_table)
    proc.start()
    data_queue = proc.get_data_queue()

    # switch to auto range
    scpi.write(b"MEAS:RANGE AUTO\n")

    # disable the output
    scpi.write(b"ROUTE:OUT 0\n")
    #scpi.write(b"ROUTE:EXT 1\n")

    print("JellyfishOPP battery profiler")
    print(f"DECIMATION_FACTOR: {DECIMATION_FACTOR}")
    print(f"DISCHARGE_CURRENT_MA: {DISCHARGE_CURRENT_MA}")
    print(f"DISCHARGE_DURATION_S: {DISCHARGE_DURATION_S}")
    print(f"REST_DURATION_S: {REST_DURATION_S}")
    print("")

    print("Connect the battery to the output terminal and press enter")

    queue_flush(data_queue)
    time.sleep(0.1)
    vout = 0
    while True:
        # read the current voltage at the output terminal
        rec = get_averaged_data(data_queue)
        vout = rec[vout_key]
        sys.stdout.write(f"\rOutput voltage: {vout/1000.0:.3f}V")
        sys.stdout.flush()
        queue_flush(data_queue)

        line = timed_input(timeout=0.2)
        if line is not None:
            print('break', line)
            break

    print("Script will now start the profiling process. Press enter to continue")
    input()

    # start with the output voltage set to the battery voltage and slowly decrease it in order to find the voltage threshold for discharge and rest currents
    try:
        vout_discharge, vout_rest = scan_thresholds(scpi, data_queue, vout)
    except KeyboardInterrupt:
        print("Interrupted")
        print("Disabling the output")
        scpi.write(b"ROUTE:OUT 0\n")
        return

    if vout_discharge==-1:
        print("ERROR: Could not find discharge current threshold, using 0V")
        vout_discharge = 0
    if vout_rest==-1:
        print("ERROR: Could not find rest current threshold, using 0V")
        vout_rest = 0

    print("Disabling the output")
    scpi.write(b"ROUTE:OUT 0\n")

    # do the profiling, save the results and calculate the battery internal resistance
    phases = [
            {
                'name': 'discharge',
                'vout': vout_discharge,
                'duration': DISCHARGE_DURATION_S,
                'output_enabled': True
                },
            {'name': 'rest',
             'vout': vout_rest,
             'duration': REST_DURATION_S,
             'output_enabled': True
             },
            ]
    phase_idx = 0
    meas = {
            'discharge': [],
            'rest': [],
            }

    try:
        while True:
            # set and enable the output
            print(f"{phases[phase_idx]['name'].upper()} phase -----------------------------")
            print(f"Setting output voltage to {int(phases[phase_idx]['vout'])}mV")
            scpi.write(b"SOURCE:VOLT %d\n"%(int(phases[phase_idx]['vout']),))
            time.sleep(0.05)
            scpi.write(b"ROUTE:OUT %d\n"%(int(phases[phase_idx]['output_enabled']),))
            st = time.time()
            while time.time()-st < phases[phase_idx]['duration']:
                # read measurements
                queue_flush(data_queue)
                time.sleep(0.1)
                rec = get_averaged_data(data_queue)
                iout = rec[iout_key]
                vout = rec[vout_key]
                print(f"\rCurrent: {iout:.3f}mA, Output voltage: {vout/1000.0:.3f}V")
                meas[phases[phase_idx]['name']].append((vout, iout))
                time.sleep(0.5)


            # calculate the battery internal resistance (after the rest phase
            if phase_idx==1:
                print("Calculating battery internal resistance")
                lr_d = meas['discharge'][-1]
                lr_r = meas['rest'][-1]
                r = (lr_r[0]-lr_d[0])/(abs(lr_d[1])-abs(lr_r[1]))
                print(f"==> Battery internal resistance: {r:.3f}Ohm")

            # switch to the next phase
            phase_idx ^= 1
    except KeyboardInterrupt:
        print("Interrupted")

    print("Disabling the output")
    scpi.write(b"ROUTE:OUT 0\n")

    proc.stop()
    return


if __name__ == "__main__":
    main()