load-controller/load_controller/bms.py

from abc import abstractmethod
import time
from dataclasses import dataclass
from enum import Enum

from .load import Load
from .temperatures import N_SENSORS

from PySide6.QtCore import QObject, Signal, Slot, Qt

from pymodbus.client import ModbusSerialClient

MODBUS_BAUDRATE = 115200
VOLTAGE_ADDRESS = 0x00
SLAVE_UNIT_ID = 0x02

NUM_CELLS = 3
VOLTAGE_QUANT = 1 / 10000.0

THRESH_UV = 2.5
THRESH_OV = 4.2
THRESH_UT = 1
THRESH_OT = 60


class BMSError(Enum):
    NONE = 0
    UV = 1
    OV = 2
    UT = 3
    OT = 4

    def __str__(self):
        if self == self.NONE:
            return ""
        elif self == self.UV:
            return "Undervoltage"
        elif self == self.OV:
            return "Overvoltage"
        elif self == self.UT:
            return "Undertemperature"
        elif self == self.OT:
            return "Overtemperature"
        else:
            return "Unknown error"


@dataclass
class BMSData:
    voltages: list[float]
    temperatures: list[float]
    error: BMSError


class BMS(QObject):
    dataUpdated = Signal(BMSData)

    _data: BMSData
    _load: Load
    _num_cells: int
    _num_sensors: int

    def __init__(self, load: Load, num_cells: int, num_sensors: int):
        super().__init__(None)
        self._load = load
        self._num_cells = num_cells
        self._num_sensors = num_sensors
        self._data = BMSData([0.0] * num_cells, [0.0] * num_sensors, BMSError.NONE)

    def _check_for_errors(self):
        error = BMSError.NONE
        for v in self._data.voltages:
            if v > THRESH_OV:
                error = BMSError.OV
                break
            elif v < THRESH_UV:
                error = BMSError.UV
                break
        for t in self._data.temperatures:
            if t > THRESH_OT:
                error = BMSError.OT
                break
            if t < THRESH_UT:
                error = BMSError.UT
                break

        if error != self._data.error:
            print(f"Error changed: {error}")
            self._data.error = error
            self._load.set_error(error != BMSError.NONE)
            self.dataUpdated.emit(self._data)

    @abstractmethod
    def do_work(self):
        pass


class BMSEvalBoard(BMS):
    _dev: ModbusSerialClient

    def __init__(self, uart_path: str, temperaturesUpdated: Signal, load: Load):
        super().__init__(load, NUM_CELLS, N_SENSORS)
        self._dev = ModbusSerialClient(
            method="rtu", port=uart_path, baudrate=MODBUS_BAUDRATE
        )
        temperaturesUpdated.connect(
            self._updateTemperatures, Qt.ConnectionType.DirectConnection
        )

    def do_work(self):
        while True:
            time.sleep(0.1)
            result = self._dev.read_holding_registers(
                VOLTAGE_ADDRESS, NUM_CELLS, SLAVE_UNIT_ID
            )
            self._data.voltages = list(
                map(lambda v: v * VOLTAGE_QUANT, result.registers)
            )
            self._check_for_errors()
            self.dataUpdated.emit(self._data)

    @Slot(list)
    def _updateTemperatures(self, temps: list[float]):
        assert len(temps) == N_SENSORS
        self._data.temperatures = temps
        self._check_for_errors()
        self.dataUpdated.emit(self._data)