Bluetooth-Support for JK BMS

buildfiles.lst

@@ -24,7 +24,9 @@ etc/dbus-serialbattery/ant.py
 etc/dbus-serialbattery/util_max17853.py
 etc/dbus-serialbattery/mnb.py
 etc/dbus-serialbattery/jkbms.py
+etc/dbus-serialbattery/jkbms_ble.py
+etc/dbus-serialbattery/jkbms_brn.py
 etc/dbus-serialbattery/sinowealth.py
 etc/dbus-serialbattery/renogy.py
 etc/dbus-serialbattery/ecs.py
-etc/dbus-serialbattery/lifepower.py
+etc/dbus-serialbattery/lifepower.py

etc/dbus-serialbattery/dbus-serialbattery.py

@@ -21,6 +21,7 @@
 from daly import Daly
 from ant import Ant
 from jkbms import Jkbms
+from jkbms_ble import Jkbms_Ble
 from sinowealth import Sinowealth
 from renogy import Renogy
 from ecs import Ecs

etc/dbus-serialbattery/installlocal.sh

@@ -1,3 +1,8 @@
 #!/bin/sh
+opkg update
+opkg install python3-misc python3-pip
+pip3 install bleak
 tar -zxf ./venus-data.tar.gz -C /data
-sh /data/etc/dbus-serialbattery/reinstalllocal.sh
+sh /data/etc/dbus-serialbattery/reinstalllocal.sh
+echo "make sure to disable Settings/Bluetooth in the Remote-Console to prevent reconnects every minute. In case of crash after ~12-16 hours disable raspberry pi 3 internal bluetooth via dtoverlay and use an external usb bluetooth-dongle"
+

etc/dbus-serialbattery/jkbms_ble.py

@@ -0,0 +1,135 @@
+# -*- coding: utf-8 -*-
+from battery import Battery, Cell
+from utils import logger
+from jkbms_brn import JkBmsBle
+from bleak import BleakScanner, BleakError
+import asyncio
+import time
+
+
+class Jkbms_Ble(Battery):
+    BATTERYTYPE = "Jkbms BLE"
+
+    def __init__(self, port, baud, address):
+        super(Jkbms_Ble, self).__init__("zero", baud)
+        self.type = self.BATTERYTYPE
+        self.jk = JkBmsBle(address)
+
+        logger.error("init of jkbmsble")
+
+    def test_connection(self):
+        # call a function that will connect to the battery, send a command and retrieve the result.
+        # The result or call should be unique to this BMS. Battery name or version, etc.
+        # Return True if success, False for failure
+
+        # check if device with given mac is found, otherwise abort
+
+        logger.info("test of jkbmsble")
+        try:
+            loop = asyncio.get_event_loop()
+            t = loop.create_task(BleakScanner.discover())
+            devices = loop.run_until_complete(t)
+        except BleakError as e:
+            logger.error(str(e))
+            return False
+
+        found = False
+        for d in devices:
+            if d.address == self.jk.address:
+                found = True
+        if not found:
+            return False
+
+        # device was found, presumeably a jkbms so start scraping
+        self.jk.start_scraping()
+        tries = 1
+
+        while self.jk.get_status() is None and tries < 20:
+            time.sleep(0.5)
+            tries += 1
+
+        # load initial data, from here on get_status has valid values to be served to the dbus
+        status = self.jk.get_status()
+        if status is None:
+            return False
+
+        if not status["device_info"]["vendor_id"].startswith("JK-"):
+            return False
+
+        logger.info("JK BMS found!")
+        return True
+
+    def get_settings(self):
+        # After successful  connection get_settings will be call to set up the battery.
+        # Set the current limits, populate cell count, etc
+        # Return True if success, False for failure
+        st = self.jk.get_status()["settings"]
+        self.cell_count = st["cell_count"]
+        self.max_battery_charge_current = st["max_charge_current"]
+        self.max_battery_discharge_current = st["max_discharge_current"]
+        self.max_battery_voltage = st["cell_ovp"] * self.cell_count
+        self.min_battery_voltage = st["cell_uvp"] * self.cell_count
+
+        for c in range(self.cell_count):
+            self.cells.append(Cell(False))
+
+        self.hardware_version = (
+            "JKBMS "
+            + self.jk.get_status()["device_info"]["hw_rev"]
+            + " "
+            + str(self.cell_count)
+            + " cells"
+        )
+        logger.info("BAT: " + self.hardware_version)
+        return True
+
+    def refresh_data(self):
+        # call all functions that will refresh the battery data.
+        # This will be called for every iteration (1 second)
+        # Return True if success, False for failure
+
+        # result = self.read_soc_data()
+        # TODO: check for errors
+        st = self.jk.get_status()
+        if st is None:
+            return False
+        if time.time() - st["last_update"] > 30:
+            # if data not updated for more than 30s, sth is wrong, then fail
+            return False
+
+        for c in range(self.cell_count):
+            self.cells[c].voltage = st["cell_info"]["voltages"][c]
+
+        self.to_temp(1, st["cell_info"]["temperature_sensor_1"])
+        self.to_temp(2, st["cell_info"]["temperature_sensor_2"])
+        self.current = st["cell_info"]["current"]
+        self.voltage = st["cell_info"]["total_voltage"]
+
+        self.soc = st["cell_info"]["battery_soc"]
+        self.cycles = st["cell_info"]["cycle_count"]
+        self.capacity = st["cell_info"]["capacity_nominal"]
+
+        # protection bits
+        # self.protection.soc_low = 2 if status["cell_info"]["battery_soc"] < 10.0 else 0
+        # self.protection.cell_imbalance = 1 if status["warnings"]["cell_imbalance"] else 0
+
+        self.protection.voltage_high = 2 if st["warnings"]["cell_overvoltage"] else 0
+        self.protection.voltage_low = 2 if st["warnings"]["cell_undervoltage"] else 0
+
+        self.protection.current_over = (
+            2
+            if (
+                st["warnings"]["charge_overcurrent"]
+                or st["warnings"]["discharge_overcurrent"]
+            )
+            else 0
+        )
+        self.protection.set_IC_inspection = (
+            2 if st["cell_info"]["temperature_mos"] > 80 else 0
+        )
+        self.protection.temp_high_charge = 2 if st["warnings"]["charge_overtemp"] else 0
+        self.protection.temp_low_charge = 2 if st["warnings"]["charge_undertemp"] else 0
+        self.protection.temp_high_discharge = (
+            2 if st["warnings"]["discharge_overtemp"] else 0
+        )
+        return True