[Python] How to read voltage, current, and battery percentage via Bluetooth from an MPPT solar...

Discussão em 'Python' iniciado por Stack, Abril 15, 2025.

  1. Stack

    Stack Membro Participativo

    I'm working on a Python project to read data from an MPPT solar controller via Bluetooth. The controller is compatible with the ZhiJinPower app (screenshot below), and it shows live telemetry such as:

    Battery voltage (e.g., 13.7V)

    Charging current (e.g., 4.4A)

    Discharge current

    Battery percentage

    Temperature

    Photovoltaic (PV) state

    Load state

    After scanning the controller with bleak, I found the following GATT services and characteristics:

    Service: 00001801-0000-1000-8000-00805f9b34fb
    Characteristic: 00002a05-0000-1000-8000-00805f9b34fb | Properties: ['indicate']

    Service: 6e400001-b5a3-f393-e0a9-e50e24dcca9e # Nordic UART
    Characteristic: 6e400003 | Properties: ['write-without-response']
    Characteristic: 6e400002 | Properties: ['notify']

    Service: 000000ff-0000-1000-8000-00805f9b34fb
    Characteristic: 0000ff02 | Properties: ['read'] ← Appears to return raw data
    Characteristic: 0000ff01 | Properties: ['read', 'write', 'notify']
    Characteristic: 0000ff04 | Properties: ['write']
    Characteristic: 0000ff03 | Properties: ['write']


    ❓ My Questions: Which characteristic is actually used to read the values like battery voltage and charging current?

    Is 0000ff02 a binary status register? I get something like: bytearray(b'\x11"3D')

    Do I need to write a command (e.g. to 6e400003) to trigger a response on 6e400002?

    What format is the data in? Is it 16-bit big-endian values that I need to scale by 100?

    I’m using Python 3.11 and the bleak library to scan and connect. Here's the screenshot from the mobile app for reference (from the official ZhiJinPower app):

    [​IMG]

    Any insight into the correct usage of these characteristics or how to decode this data properly would be hugely appreciated!

    code snippet:

    import asyncio
    import subprocess
    from bleak import BleakClient, BleakScanner
    from datetime import datetime
    import csv
    import traceback

    # BLE device MAC address and characteristic UUID
    DEVICE_ADDRESS = "34:B7:DA:86:B2:D6"
    STATUS_UUID = "0000ff01-0000-1000-8000-00805f9b34fb"

    # CSV file to log the data
    CSV_FILE = "mppt_status_log.csv"

    # Decode MPPT status from raw bytearray (extended to match app UI)
    def decode_mppt_status(data: bytearray) -> dict:
    try:
    return {
    "Timestamp": datetime.now().isoformat(),
    "Temperature (°C)": data[0] / 10,
    "Charging Current (A)": data[1] / 10,
    "Discharge Current (A)": data[2] / 10,
    "Battery Voltage (V)": int.from_bytes(data[3:5], 'little') / 100,
    "Battery %": data[5] / 1,
    "Accumulated Power (Ah)": int.from_bytes(data[6:8], 'little') / 10,
    "Wind State": "Open" if data[8] else "Close",
    "Photovoltaic State": "Daytime" if data[9] else "Night",
    "Load State": "Open" if data[10] else "Close"
    }
    except Exception as e:
    return {"Timestamp": datetime.now().isoformat(), "error": f"Decode failed: {e}"}

    # Save a single row to CSV
    def save_to_csv(data: dict):
    write_header = False
    try:
    with open(CSV_FILE, "r") as f:
    pass
    except FileNotFoundError:
    write_header = True

    with open(CSV_FILE, "a", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=data.keys())
    if write_header:
    writer.writeheader()
    writer.writerow(data)

    # Restart Bluetooth service if needed
    def restart_bluetooth_service():
    print(" Restarting Bluetooth service...")
    subprocess.run(["sudo", "systemctl", "restart", "bluetooth"], check=False)

    # Optional: verify device is visible before connecting
    async def device_is_visible(mac):
    print(" Checking if device is visible...")
    devices = await BleakScanner.discover()
    return any(d.address.upper() == mac.upper() for d in devices)

    # Main loop with retry and timeout patch
    async def read_status_loop(interval=10):
    try:
    if not await device_is_visible(DEVICE_ADDRESS):
    print("❌ Device not advertising. Ensure it is powered and nearby.")
    return

    async with BleakClient(DEVICE_ADDRESS, timeout=20.0) as client:
    if client.is_connected:
    print(f"\n✅ Connected to {DEVICE_ADDRESS}\nLogging data every {interval}s...\n")
    while True:
    try:
    raw = await client.read_gatt_char(STATUS_UUID)
    print(f"\n Raw bytes: {raw}")
    print(f" Decoded as integers: {list(raw)}")

    interpreted = decode_mppt_status(raw)
    for k, v in interpreted.items():
    print(f"{k}: {v}")

    print("\n" + "="*40 + "\n")
    save_to_csv(interpreted)
    except Exception as e:
    print(f"❌ Read error: {e}")

    await asyncio.sleep(interval)
    else:
    print("❌ Failed to connect to device.")
    except Exception as e:
    print("❌ Connection error:")
    traceback.print_exc()
    restart_bluetooth_service()

    print(" Retrying connection in 10 seconds...")
    await asyncio.sleep(10)
    await read_status_loop(interval)

    if __name__ == "__main__":
    asyncio.run(read_status_loop(interval=10))

    Continue reading...

Compartilhe esta Página