Liontron 200AH Batterie

Ich habe versucht die Liontron über VanOs mit Hilfe von phyton3 pexpect einzubinden = Keine Werte, das Script liontronBmsInfo.py baut auch keine Verbindung über die MAC Adresse auf.

Hoffentlich hat einer von euch eine Lösung

wie immer das problem war vor dem Bildschirm, hatte nicht gesehen das es bei der Liontron eine Mac mit UUID gab. danach funzzte es

ich habe mein Project für einige Zeit geparkt und nun PeKaWay neu aufgesetzt in der Version **v2.0.8 .. Nun will meine Liontron 200 AH sich nicht mehr verbinden … ich hatte dort über die Liontron App einen Treiber - Update gemacht und die Batterie mit einem PIN (6stellen) gesichert. … ich benötige TIPS

Da wird wahrscheinlich so schnell nichts kommen, da Liontron seit neusten eine Komplett eigene Firmware auf das JBD Bms schreibt. Da ist abzuwarten ob Pekaway von Liontron das Protokoll bekommt um diese auslesen zu können.

Gruß Wulle

Vielen Dank für die Info, Habe mit Liontron gesprochen und sie arbeiten an einem Treiber für Victron ‘Liontron EkoCast’ - Treiber … habe mich auf die BetaTest -Liste setzen lassen. Dann ist auch die Lösung für PeKaWay nicht so weit

Gibt es hier eigentlich einen neuen Stand?
Seit dem Firmware-update auf 0.62 meiner Liontron 200AH kann sie nicht mehr von Pekaway ausgelesen werden und hier gibt es inzwischen ja ein paar infos.

Moin,

ich hab das mal in die KI geworfen und das jbd_bms_reader2.py umschreiben lassen.
Ich hab keine Batterie da zum testen.
Eigentlich musst du nur den Datei Inhalt ersetzen, dann sollte es funktionieren.

#!/usr/bin/env python3
"""Continuously read LIONTRON EKOcast broadcasts using the legacy Node-RED JSON schema.

Usage: python3 jbd_bms_reader2.py -d AA:BB:CC:DD:EE:FF [-v|-vv]
Requires bleak. See LIONTRON_EKOCAST_PROTOKOLL.md for field limitations.
"""
import argparse
import asyncio
import json
import os
import re
import sys

MANUFACTURER_ID = 0x0211
RETRY_SECONDS = 5
WAIT_SECONDS = 30
PUBLISH_SECONDS = 1.0


def parse_payload(raw):
    """Decode the 27-byte payload WITHOUT the two company-ID bytes."""
    if len(raw) >= 60 and raw[0] == 0x01:
        raise ValueError("BLE5-Extended-Frame: im gelieferten Treiber nicht dokumentiert")
    if len(raw) != 27:
        raise ValueError("Erwartet: 27 Bytes Manufacturer-Payload, erhalten: %d" % len(raw))

    def u16(offset):
        return (raw[offset] << 8) | raw[offset + 1]

    remaining = u16(8) / 100.0
    nominal = u16(10) / 100.0
    temperature = ((((raw[13] & 0xF8) >> 3) | ((raw[12] & 1) << 5)) * 2) - 30
    # Preserve the old SoC calculation; avoid division by zero.
    soc = round(remaining / nominal * 100.0, 2) if nominal else (raw[12] >> 1)
    data = {
        "Vmain": (u16(4) >> 3) / 100.0,
        "Imain": (u16(6) - 32767) / 100.0,
        "RemainAh": remaining,
        "NominalAh": nominal,
        "SoC": soc,
        # Compatibility alias, NOT a separate measured MOS temperature.
        "TempMOS": float(temperature),
        "TempC1": None,
        "TempC2": None,
        "NumberCycles": (raw[14] << 6) | (raw[15] >> 2),
        # No documented protection bitmap in the supplied broadcast parser.
        "ProtectState": None,
        "ProtectStateBin": None,
        "ProtectStateText": "unknown",
    }
    for cell in range((raw[13] & 7) + 1):
        data["Vcell%d" % (cell + 1)] = (raw[17 + cell] * 10 + 1500) / 1000.0
    return data


def mac_address(value):
    if not re.fullmatch(r"(?:[0-9a-fA-F]{2}:){5}[0-9a-fA-F]{2}", value):
        raise argparse.ArgumentTypeError("MAC muss das Format AA:BB:CC:DD:EE:FF haben")
    return value.upper()


def load_scanner():
    # Reuse an installed EKOcast copy without importing its service/driver.
    for path in (os.path.join(os.path.dirname(os.path.abspath(__file__)), "ext"),
                 "/data/lt_ekocast/ext"):
        if os.path.isdir(path):
            sys.path.append(path)
    from bleak import BleakScanner
    return BleakScanner


async def read_continuously(address, scanner_factory, verbose=0):
    """Yield fresh broadcasts indefinitely without connecting via GATT."""
    def log(message):
        if verbose:
            print(message, file=sys.stderr, flush=True)

    passive = sys.platform == "linux"
    while True:
        pending = asyncio.Queue(maxsize=1)
        last_publish = float("-inf")
        rejected = set()

        def detected(device, advertisement):
            if device.address.upper() != address:
                return
            raw = advertisement.manufacturer_data.get(MANUFACTURER_ID)
            if raw is None:
                return
            try:
                data = parse_payload(raw)
            except ValueError as exc:
                reason = str(exc)
                if reason not in rejected:
                    log("BMS %s: %s; warte auf passendes Paket" % (address, reason))
                    rejected.add(reason)
                return
            if verbose >= 2:
                log("BMS Manufacturer-Payload: " + bytes(raw).hex())
            if pending.full():
                pending.get_nowait()
            pending.put_nowait(data)

        scanner = None
        started = False
        try:
            options = {"detection_callback": detected}
            if passive:
                options.update(scanning_mode="passive", bluez={
                    "or_patterns": [(0, 0xFF, b"\x11\x02")],
                })
            else:
                options.update(scanning_mode="active")
                if sys.platform == "linux":
                    options["bluez"] = {"filters": {"DuplicateData": True}}
            scanner = scanner_factory(**options)
            await asyncio.wait_for(scanner.start(), timeout=20)
            started = True
            log("BMS wartet auf EKOcast-Broadcast von " + address)
            # Keep scanning when absent; do not repeatedly reset the adapter.
            while True:
                try:
                    data = await asyncio.wait_for(pending.get(), timeout=WAIT_SECONDS)
                except asyncio.TimeoutError:
                    log("BMS keine neuen Daten; suche weiter nach " + address)
                    continue
                now = asyncio.get_running_loop().time()
                if now - last_publish >= PUBLISH_SECONDS:
                    last_publish = now
                    yield data
        except Exception as exc:
            log("BMS Scanfehler: %s; neuer Versuch in %ds" % (exc, RETRY_SECONDS))
            if passive and not started:
                # Older BlueZ may not support AdvertisementMonitor.
                # Active discovery also receives broadcasts without connecting.
                passive = False
                log("BMS verwende BLE-Discovery ohne Verbindungsaufbau")
        finally:
            if scanner is not None:
                try:
                    await asyncio.wait_for(scanner.stop(), timeout=10)
                except Exception as exc:
                    log("BMS Scanner stoppen: %s" % exc)
        await asyncio.sleep(RETRY_SECONDS)


async def stream(address, scanner_factory, verbose=0):
    async for data in read_continuously(address, scanner_factory, verbose):
        # One complete JSON object per line for Node-RED exec in spawn mode.
        print(json.dumps(data, sort_keys=False), flush=True)


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("-d", "--device", type=mac_address, metavar="MAC", required=True,
                        help="Specify remote Bluetooth address")
    parser.add_argument("-v", "--verbose", dest="v", action="count", default=0,
                        help="Verbosity (diagnostics on stderr)")
    args = parser.parse_args(argv)
    try:
        scanner = load_scanner()
    except ImportError as exc:
        print("BMS: BLE-Abhaengigkeit fehlt (%s). bleak im verwendeten Python "
              "installieren; siehe LIONTRON_EKOCAST_PROTOKOLL.md." % exc, file=sys.stderr)
        return 1
    try:
        asyncio.run(stream(args.device, scanner, args.v))
    except KeyboardInterrupt:
        return 130
    return 0


if __name__ == "__main__":
    sys.exit(main())