MODBUS_Example.py

#!/usr/bin/env python3
'''
This script depends on the following Victron equipment. BMV, Multiplus with ESS, Solar Charger, Venus GX device.
Modbus must be enabled in the Venus GX device
You shouldnt have to change anything but some variables to make this work with your system
provided you actually have the requisite victron equipment.
'''

#from pymodbus.client import ModbusTcpClient as ModbusClient
#from pymodbus.utilities import convert_from_registers

from pymodbus.client import ModbusTcpClient as ModbusClient
from datetime import datetime
import sys
import os
import time
import textwrap
import subprocess

'''
==========================================================
                 Change the Variables Below               
==========================================================
'''
Analog_Inputs = 'n'  # Y or N (case insensitive) to display Gerbo GX Analog Temperature inputs
ESS_Info      = 'n'  # Y or N (case insensitive) to display ESS system information
ip            = "192.168.20.156"  # ip address of GX device or if on venus local try localhost

# Value Refresh Rate in seconds
RefreshRate = 1

# Unit ID #'s from Cerbo GX.
# Do not confuse UnitID with Instance ID.
# You can also get the UnitID from the GX device. Menu --> Settings --> Services --> ModBusTCP --> Available Services
# Or
# https://github.com/victronenergy/dbus_modbustcp/blob/master/CCGX-Modbus-TCP-register-list.xlsx Tab #2
#===================================
SolarChargerID    = 226
MultiPlusID       = 227
BmvID             = 223
VEsystemID        = 100
TempSensor1_ID    = 24
TempSensor2_ID    = 25
TempSensor3_ID    = 26
# Temp Sensors Name
Sens1 = "Battery Box"
Sens2 = "Cabin"
Sens3 = "Outside"
'''
==========================================================
                Change the Variables Above                
==========================================================
'''
#===================================
tr = textwrap.TextWrapper(width=56, subsequent_indent=" ")
print("\033[H\033[J")  # Clear screen
print('\033[?25l', end="")  # Hide Blinking Cursor
clear = "\033[K\033[1K"  # Eliminates screen flashing / blink during refresh

# Local network ip address of Cerbo GX. Default port 502
client = ModbusClient(ip, port='502')

# All modbus variable requests are sent to this function and it return's the requested value
def modbus_register(address, unit, data_type="int16"):
    msg = client.read_holding_registers(address, count=1, device_id=unit)
    if not msg.isError():
        value = msg.registers[0]  # Extract the first (and only) 16-bit register value
        if data_type == "int16":
            # Convert unsigned 16-bit to signed if value > 32767
            if value > 32767:
                value -= 65536
        # For uint16, no conversion needed
        return value
    else:
        raise AttributeError(f"Modbus read error at address {address}, device_id {unit}")

#===================================
# Module for standard text colors, foreground and background
class colors:
    reset         = '\033[0m'
    bold          = '\033[01m'
    disable       = '\033[02m'
    underline     = '\033[04m'
    reverse       = '\033[07m'
    strikethrough = '\033[09m'
    invisible     = '\033[08m'
    blink         = '\033[05m'

    class fg:
        red          = '\033[38;5;1m'
        light_red    = '\033[38;5;9m'
        cyan         = '\033[38;5;6m'
        light_cyan   = '\033[38;5;14m'
        gray         = '\033[38;5;240m'
        light_gray   = '\033[38;5;246m'
        white        = '\033[38;5;15m'
        black        = '\033[38;5;16m'
        orange       = '\033[38;5;202m'
        light_orange = '\033[38;5;172m'
        blue         = '\033[38;5;21m'
        light_blue   = '\033[38;5;39m'
        green        = '\033[38;5;28m'
        light_green  = '\033[38;5;34m'
        purple       = '\033[38;5;93m'
        light_purple = '\033[38;5;99m'
        yellow       = '\033[38;5;220m'
        light_yellow = '\033[38;5;227m'
        pink         = '\033[38;5;201m'
        light_pink   = '\033[38;5;206m'

    class bg:
        red          = '\033[48;5;1m'
        light_red    = '\033[48;5;9m'
        cyan         = '\033[48;5;6m'
        light_cyan   = '\033[48;5;14m'
        gray         = '\033[48;5;240m'
        light_gray   = '\033[48;5;246m'
        white        = '\033[48;5;15m'
        black        = '\033[48;5;16m'
        orange       = '\033[48;5;202m'
        light_orange = '\033[48;5;172m'
        blue         = '\033[48;5;21m'
        light_blue   = '\033[48;5;39m'
        green        = '\033[48;5;28m'
        light_green  = '\033[48;5;34m'
        purple       = '\033[48;5;93m'
        light_purple = '\033[48;5;99m'
        yellow       = '\033[48;5;220m'
        light_yellow = '\033[48;5;227m'

#===================================
SolarStateDict = {0:  "OFF",
                  2:  "Fault",
                  3:  "Bulk",
                  4:  "Absorption",
                  5:  "Float",
                  6:  "Storage",
                  7:  "Equalize",
                  11: "Other Hub-1",
                  245: "Wake-Up",
                  252: "EXT Control"}

ESSbatteryLifeStateDict = {0: "Battery Life Disabled",
                           1: "Restarting",
                           2: "Self-consumption",
                           3: "Self consumption, SoC exceeds 85%",
                           4: "Self consumption, SoC at 100%",
                           5: "Discharge Disabled. SoC below BatteryLife Dynamic SoC",
                           6: "SoC has been below SoC limit for more than 24 hours. Slow Charging battery",
                           7: "Multi is in sustain mode",
                           8: "Recharge, SOC dropped 5% or more below MinSOC",
                           9: "Keep batteries charged mode enabled",
                           10: "Self consumption, SoC at or above minimum SoC",
                           11: "Discharge Disabled (Low SoC), SoC is below minimum SoC",
                           12: "Recharge, SOC dropped 5% or more below minimum"
                           }

VEbusStatusDict =  {0:  "OFF",
                    1:  "Low Power",
                    2:  "Fault",
                    3:  "Bulk Charging",
                    4:  "Absorption Charging",
                    5:  "Float Charging",
                    6:  "Storage",
                    7:  "Equalize",
                    8:  "Passthru",
                    9:  "Inverting",
                    10: "Power Assist",
                    11: "Power Supply Mode",
                    246: "Repeated Absorption",
                    247: "Equalize",
                    248: "Battery Safe",
                    249: "Test",
                    250: "Blocked",
                    251: "Test",
                    252: "External control",
                    256: "Discharging",
                    257: "Sustain",
                    258: "Recharging",
                    259: "Scheduled Charge"
                    }

VEbusErrorDict = {
                  0: "No Error",
                  1: "Error 1: Device is switched off because one of the other phases in the system has switched off",
                  2: "Error 2: New and old types MK2 are mixed in the system",
                  3: "Error 3: Not all- or more than- the expected devices were found in the system",
                  4: "Error 4: No other device whatsoever detected",
                  5: "Error 5: Overvoltage on AC-out",
                  6: "Error 6: in DDC Program",
                  7: "VE.Bus BMS connected- which requires an Assistant- but no assistant found",
                  8: "Error 8: Ground Relay Test Failed",
                  9: "VE.Bus Error 9",
                  10: "VE.Bus Error 10: System time synchronisation problem occurred",
                  11: "Error 11: Relay Test Fault - Installation error or possibly relay failure",
                  12: "Error 12: - Config mismatch with 2nd mcu",
                  13: "VE.Bus Error 13",
                  14: "Error 14: Device cannot transmit data",
                  15: "Error 15 - VE.Bus combination error",
                  16: "Error 16: Dongle missing",
                  17: "Error 17: One of the devices assumed master status because the original master failed",
                  18: "Error 18: AC Overvoltage on the output of a slave has occurred while already switched off",
                  19: "Error 19 - Slave does not have AC input!",
                  20: "Error 20: - Configuration mismatch",
                  21: "VE.Bus Error 21",
                  22: "Error 22: This device cannot function as slave",
                  23: "VE.Bus Error 23",
                  24: "Error 24: Switch-over system protection initiated",
                  25: "Error 25: Firmware incompatibility. The firmware of a connected device is not sufficiently up to date.",
                  26: "Error 26: Internal error",
                  27: "VE.Bus Error 27",
                  28: "VE.Bus Error 28",
                  29: "VE.Bus Error 29",
                  30: "VE.Bus Error 30",
                  31: "VE.Bus Error 31",
                  32: "VE.Bus Error 32"
                  }

def spacer():
    print(colors.fg.gray, "="*80, sep="")

print(f"Test")
while True:
    print("\033[0;0f")  # move to col 0 row 0
    screensize = os.get_terminal_size()
    # Modbus variable requests
    try:
        BatterySOC    = modbus_register(266, BmvID) / 10
        BatteryWatts  = modbus_register(842, VEsystemID)
        BatteryAmps   = modbus_register(841, VEsystemID) / 10
        BatteryVolts  = modbus_register(259, BmvID) / 100

        SolarVolts    = modbus_register(776, SolarChargerID) / 100
        SolarAmps     = modbus_register(777, SolarChargerID) / 10
        SolarWatts    = modbus_register(789, SolarChargerID) / 10
        MaxSolarWatts = modbus_register(785, SolarChargerID)
        SolarYield    = modbus_register(784, SolarChargerID) / 10
        SolarState    = modbus_register(775, SolarChargerID)

        GridSetPoint  = modbus_register(2700, VEsystemID)
        GridCondition = modbus_register(64, MultiPlusID)
        ACoutHZ       = modbus_register(21, MultiPlusID) / 100
        ACoutVolts    = modbus_register(15, MultiPlusID) / 10
        ACoutAmps     = modbus_register(18, MultiPlusID) / 10
        ACoutWatts    = modbus_register(817, VEsystemID)
        GridHZ        = modbus_register(9, MultiPlusID) / 100
        GridVolts     = modbus_register(3, MultiPlusID) / 10
        GridAmps      = modbus_register(6, MultiPlusID) / 10
        GridWatts     = modbus_register(820, VEsystemID)

        ESSsocLimitUser     = modbus_register(2901, VEsystemID) / 10
        ESSsocLimitDynamic  = modbus_register(2903, VEsystemID) / 10
        ESSbatteryLifeState = modbus_register(2900, VEsystemID)

        VEbusError  = modbus_register(32, MultiPlusID)
        VEbusStatus = modbus_register(31, MultiPlusID)

        if Analog_Inputs.lower() == "y":
            try:
                TempSensor1 = modbus_register(3304, TempSensor1_ID) / 100 * 1.8 + 32
            except AttributeError:
                TempSensor1 = 777

            try:
                TempSensor2 = modbus_register(3304, TempSensor2_ID) / 100 * 1.8 + 32
            except AttributeError:
                TempSensor2 = 777

            try:
                TempSensor3 = modbus_register(3304, TempSensor3_ID) / 100 * 1.8 + 32
            except AttributeError:
                TempSensor3 = 777

        # Datetime object containing current date and time
        now = datetime.now()

        # Fri 21 Jan 2022 09:06:57 PM
        dt_string = now.strftime("%a %d %b %Y %r")

        print(clear, colors.fg.purple, f"\n Time & Date............. {dt_string}", sep="")

        #===========================================================================================
        # Display Section
        # Battery value color
        if BatterySOC >= 60:
            BatteryColor = colors.fg.green
        elif 30 <= BatterySOC < 60:
            BatteryColor = colors.fg.yellow
        else:
            BatteryColor = colors.fg.red
        print(clear, colors.fg.cyan, f" Battery SOC............. ", colors.bold, BatteryColor, f"{BatterySOC:.1f}", " %", colors.reset, sep="")
        print(clear, colors.fg.cyan, f" Battery Watts........... {BatteryWatts:.0f}", sep="")
        print(clear, colors.fg.cyan, f" Battery Amps............ {BatteryAmps:.1f}", sep="")
        print(clear, colors.fg.cyan, f" Battery Volts........... {BatteryVolts:.2f}", colors.reset, sep="")
        spacer()

        print(clear, colors.fg.orange, f" PV Watts................ {SolarWatts:.0f}", sep="")
        print(clear, f" PV Amps................. {SolarAmps:.2f}", sep="")
        print(clear, f" PV Volts................ {SolarVolts:.2f}", sep="")
        print(clear, f" Max PV Watts Today...... {MaxSolarWatts}", sep="")
        print(clear, f" PV Yield Today.......... {SolarYield:.3f} kWh", sep="")
        print(clear, f" PV Charger State........ {SolarStateDict[SolarState]}", sep="")
        spacer()

        print(clear, colors.fg.green, f" Grid Set Point Watts.... {GridSetPoint}", sep="")
        print(clear, f" Grid Watts.............. {GridWatts:.0f}\t\tAC Output Watts......... {ACoutWatts}", sep="")
        print(clear, f" Grid Amps............... {GridAmps:.1f}\t\tAC Output Amps.......... {ACoutAmps:.1f}", sep="")
        print(clear, f" Grid Volts ............. {GridVolts:.1f}\t\tAC Output Volts......... {ACoutVolts:.1f}", sep="")
        print(clear, f" Grid Freq .............. {GridHZ:.1f}\t\tAC Output Freq.......... {ACoutHZ:.1f}", sep="")

        if GridCondition == 0:
            GC = "OK"
            GC_Color = colors.fg.green
        else:
            GC = "Grid LOST"
            GC_Color = colors.fg.light_red
        print(clear, f"{GC_Color} Grid Condition.......... {GC}", sep="")
        spacer()

        # VE.Bus Status
        print(clear, colors.fg.light_blue, end="", sep="")
        if VEbusStatus == 2:
            VEbusStatus_Color = colors.fg.red
        else:
            VEbusStatus_Color = colors.fg.light_blue
        print(clear, f" System State............ {VEbusStatus_Color}{VEbusStatusDict[VEbusStatus]}", sep="")

        # VE.Bus Error
        print(clear, colors.fg.light_blue, end="", sep="")
        if VEbusError == 0:
            print(clear, f" VE.Bus Error............ ", colors.fg.green, "No Error", sep="")
        else:
            print(clear, f" VE.Bus Error............ ", colors.fg.red, tr.fill(f"{VEbusErrorDict[VEbusError]}"), "\033[K", sep="")

        # ESS Info
        if ESS_Info.lower() == "y":
            if 1 <= ESSbatteryLifeState <= 8:
                print(clear, colors.fg.light_blue, f" ESS SOC Limit (User).... {ESSsocLimitUser:.0f}% Unless Grid Fails ", sep="")
                print(clear, colors.fg.light_blue, f" ESS SOC Limit (Dynamic). {ESSsocLimitDynamic:.0f}%", sep="")
                print(clear, colors.fg.light_blue, f" ESS Mode ............... Optimized (With Battery Life)", sep="")
            elif ESSbatteryLifeState == 9:
                print(clear, colors.fg.light_blue, f" ESS Mode................ Keep Batteries Charged Mode Enabled", sep="")
            elif 10 <= ESSbatteryLifeState <= 12:
                print(clear, colors.fg.light_blue, f" ESS SOC Limit (User).... {ESSsocLimitUser:.0f}% Unless Grid Fails", sep="")
                print(clear, colors.fg.light_blue, f" ESS Mode ............... Optimized (Without Battery Life)", sep="")
            if ESSbatteryLifeState != 9:
                print(clear, colors.fg.light_blue, f" ESS Battery State....... {ESSbatteryLifeStateDict[ESSbatteryLifeState]}", sep="")

        # Cerbo GX Analog Temperature Inputs
        spacer()
        if Analog_Inputs.lower() == "y":
            if TempSensor1 == 777:
                print(clear, colors.fg.pink, f" Temp Sensor 1........... Not installed or unit ID wrong", sep="")
            elif TempSensor1 >= 50:
                print(clear, colors.fg.pink, f" {Sens1} Temp........ {TempSensor1:.1f} °F", colors.fg.red, " Whew...its a tad warm in here", sep="")
            else:
                print(clear, colors.fg.pink, f" {Sens1} Temp........ {TempSensor1:.1f} °F ", sep="")

            if TempSensor2 == 777:
                print(clear, colors.fg.pink, " Temp Sensor 2........... Not installed or unit ID wrong", sep="")
            elif TempSensor2 < 40:
                print(clear, colors.fg.pink, f" {Sens2} Temp.............. {TempSensor2:.1f} °F", colors.fg.blue, " Whoa!..Crank up the heat in this place!", sep="")
            else:
                print(clear, colors.fg.pink, f" {Sens2} Temp.............. {TempSensor2:.1f} °F", sep="")

            if TempSensor3 == 777:
                print(clear, colors.fg.pink, " Temp Sensor 3........... Not installed or unit ID wrong", colors.reset, sep="")
            elif TempSensor3 < 33:
                print(clear, colors.fg.pink, f" {Sens3} Temp............ {TempSensor3:.1f} °F", colors.fg.blue, " Burr...A Wee Bit Chilly Outside", colors.reset, sep="")
            else:
                print(clear, colors.fg.pink, f" {Sens3} Temp............ {TempSensor3:.1f} °F", colors.reset, sep="")

        print(clear, colors.fg.gray, "\n\tCtrl+C To Quit", colors.reset)
        time.sleep(RefreshRate)
        if screensize != os.get_terminal_size():
            print("\033[H\033[J")  # Clear screen

    except KeyboardInterrupt:
        print(clear)
        print("\033[J")
        print(colors.reset)
        print('\033[?25h', end="")  # Restore Blinking Cursor
        client.close()
        quit()

    except AttributeError:
        continue