calibration_tool.py

"""
Automated sensor calibration tool.
Sends calibration commands over UART, reads back raw ADC values,
computes gain/offset coefficients, and writes them to device NVS flash.

Usage:
    python calibration_tool.py COM3 --points 5 --channel 0
"""
import argparse
import struct
import time
import csv
import os
from serial_reader import SerialReader

CAL_START_CMD  = bytes([0xAA, 0x01, 0x10, 0x11])  # cmd: start calibration
CAL_SAMPLE_CMD = bytes([0xAA, 0x01, 0x11, 0x10])  # cmd: take sample
CAL_WRITE_CMD  = bytes([0xAA, 0x05, 0x12])         # cmd: write coefficients (payload follows)
CAL_RESPONSE   = 0x20                               # expected response type


def least_squares(points: list[tuple]) -> tuple[float, float]:
    """Simple linear regression — returns (gain, offset)."""
    n = len(points)
    sx  = sum(p[0] for p in points)
    sy  = sum(p[1] for p in points)
    sxy = sum(p[0] * p[1] for p in points)
    sxx = sum(p[0] ** 2 for p in points)
    denom = n * sxx - sx ** 2
    if abs(denom) < 1e-9:
        raise ValueError("Calibration points are collinear — check inputs")
    gain   = (n * sxy - sx * sy) / denom
    offset = (sy - gain * sx) / n
    return gain, offset


def read_adc_sample(reader: SerialReader, timeout=3.0) -> int | None:
    deadline = time.time() + timeout
    while time.time() < deadline:
        item = reader.get(timeout=0.05)
        if item:
            _, raw = item
            if len(raw) >= 6 and raw[0] == 0xAA and raw[2] == CAL_RESPONSE:
                return struct.unpack('<H', raw[3:5])[0]
    return None


def run_calibration(port, baud, num_points, channel):
    reader = SerialReader(port, baud)
    reader.start()
    points = []

    print(f"\n=== Calibration: CH{channel} | {num_points} points ===\n")
    for i in range(num_points):
        ref = float(input(f"  Point {i+1}/{num_points} — Enter reference value: "))
        reader.send(CAL_SAMPLE_CMD)
        raw_adc = read_adc_sample(reader)
        if raw_adc is None:
            print("  [!] No response from device — aborting")
            reader.stop()
            return
        print(f"  ADC raw: {raw_adc}  |  Reference: {ref}")
        points.append((raw_adc, ref))

    gain, offset = least_squares(points)
    print(f"\nGain:   {gain:.6f}")
    print(f"Offset: {offset:.6f}")

    payload = struct.pack('<Bff', channel, gain, offset)
    cmd = CAL_WRITE_CMD + payload + bytes([sum(payload) & 0xFF])
    reader.send(cmd)
    print("Coefficients written to device NVS.")

    log_path = f"cal_ch{channel}_{int(time.time())}.csv"
    with open(log_path, 'w', newline='') as f:
        w = csv.writer(f)
        w.writerow(['raw_adc', 'reference', 'gain', 'offset'])
        for raw, ref in points:
            w.writerow([raw, ref, gain, offset])
    print(f"Calibration log saved: {log_path}")
    reader.stop()


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Embedded sensor calibration tool')
    parser.add_argument('port')
    parser.add_argument('--baud',   type=int, default=115200)
    parser.add_argument('--points', type=int, default=3)
    parser.add_argument('--channel', type=int, default=0)
    args = parser.parse_args()
    run_calibration(args.port, args.baud, args.points, args.channel)