import dbus import time import logging import xml.etree.ElementTree as ET logger = logging.getLogger('main') #dbus paths path_setpoint= '/Hub4/L1/AcPowerSetpoint' path_power= '/Ac/ActiveIn/L1/P' path_bat_max_chargecurrent= '/BatteryOperationalLimits/MaxChargeCurrent' path_bat_current= '/Dc/0/Current' path_bat_power= '/Dc/0/Power' path_disable_feedin ='/Hub4/DisableFeedIn' path_disable_charge ='/Hub4/DisableCharge' path_socvalue = '/Soc' path_batt_socvalue = '/Soc' path_batt_current = '/Dc/0/Current' path_batt_power = '/Dc/0/Power' path_batt_temperature = '/Dc/0/Temperature' path_batt_voltage_sum = '/Dc/0/Voltage' path_batt_voltage_lowest = '/System/MinCellVoltage' path_batt_voltage_highest = '/System/MaxCellVoltage' path_batt_voltage_diff = '/Voltages/Diff' class client: def __init__(self) -> None: self.setpoint = 0 self.soc = None self.power_usage = None self.session_bus = dbus.SystemBus() self.battery_power = None self.battery_current = None self.battery_max_chargecurrent = None self.batteries = [] self.locate_device() def __getPaths(self, object, current_path): objectManager = self.session_bus.get_object(object, '/' if current_path == '' else current_path) introspection_interface = dbus.Interface( objectManager, dbus.INTROSPECTABLE_IFACE, ) tree = ET.ElementTree(ET.fromstring(introspection_interface.Introspect())) retval = [] root = tree.getroot() for node in root: if node.tag =='node': print (current_path + '/' + node.attrib['name']) for path in self.__getPaths(object,current_path + '/' + node.attrib['name'] ): retval.append(current_path + node.attrib['name'] + '/' +path ) return retval def listPaths(self, object): print(self.__getPaths(object,'')) #return self.__getPaths(object,'/') def locate_device(self): logger.info("Locating victron devices") for name in self.session_bus.list_names(): if name[0:17] == "com.victronenergy": parts = name.split(".") if parts[2]=='vebus': print("Detected vbus {}".format(name)) self.device = name if parts[2]=='battery': batteryname = self.session_bus.call_blocking(name, "/Serial",None,"GetValue","",[]) print("Detected battery {} ({})".format(name,batteryname)) self.batteries.append(battery(self,name,batteryname)) def set_charging(self): self.session_bus.call_blocking(self.device, path_disable_feedin ,None,'SetValue', 'v', [1]) self.session_bus.call_blocking(self.device, path_disable_charge ,None,'SetValue', 'v', [0]) def set_idle(self): self.session_bus.call_blocking(self.device, path_disable_feedin ,None,'SetValue', 'v', [1]) self.session_bus.call_blocking(self.device, path_disable_charge ,None,'SetValue', 'v', [1]) def set_discharging(self): self.session_bus.call_blocking(self.device, path_disable_feedin ,None,'SetValue', 'v', [0]) self.session_bus.call_blocking(self.device, path_disable_charge ,None,'SetValue', 'v', [0]) def connect(self): while True: try: logger.info("Connecting to dbus...") logger.info("Return from setpoint: {}".format(self.session_bus.call_blocking(self.device, path_setpoint ,None,'SetValue', 'v', [self.setpoint]))) break except: logger.info("Unable to connect to dbus {}".format(path_setpoint)) time.sleep(5) def get_power_usage(self): return self.power_usage def get_soc(self): return self.soc def get_setpoint(self): return self.setpoint def set_setpoint(self, setpoint): try: self.session_bus.call_blocking(self.device, path_setpoint ,None,'SetValue', 'v', [setpoint]) self.setpoint = setpoint except: logger.info("Unable to connect to dbus {}".format(path_setpoint)) self.setpoint = None def get_battery_current(self): return self.battery_current def get_battery_power(self): return self.battery_power def get_battery_max_chargecurrent(self): return self.battery_max_chargecurrent def update(self): logger.debug("Run victron update") self.power_usage = self.session_bus.call_blocking(self.device, path_power ,None,'GetValue', '', []) if not isinstance(self.power_usage, int): logger.info('Power usage is not an int value') self.power_usage = None return self.soc = self.session_bus.call_blocking(self.device, path_socvalue ,None,'GetValue', '', []) if not isinstance(self.soc, float): logger.info('SoC is not a float value') self.soc = None return self.setpoint = self.session_bus.call_blocking(self.device, path_setpoint ,None,'GetValue', '', []) self.battery_power = self.session_bus.call_blocking(self.device, path_bat_power ,None,'GetValue', '', []) self.battery_current = self.session_bus.call_blocking(self.device, path_bat_current ,None,'GetValue', '', []) self.battery_max_chargecurrent = self.session_bus.call_blocking(self.device, path_bat_max_chargecurrent ,None,'GetValue', '', []) for battery in self.batteries: battery.update() class battery: def __init__(self, victron: client, device: str, name: str) -> None: self.soc = None self.current = None self.power = None self.temperature = None self.voltage_sum = None self.voltage_lowest = None self.voltage_highest = None self.voltage_diff = None self.name = name self.device = device self.victron = victron #self.paths = victron.listPaths(device) def get_name(self): return self.snameoc def get_soc(self): return self.soc def get_current(self): return self.current def get_power(self): return self.power def get_temperature(self): return self.temperature def get_voltage_sum(self): return self.voltage_sum def get_voltage_lowest(self): return self.voltage_lowest def get_voltage_highest(self): return self.voltage_highest def get_voltage_diff(self): return self.voltage_diff def update(self): self.soc = self.victron.session_bus.call_blocking(self.device,path_batt_socvalue,None,'GetValue', '', []) self.current = self.victron.session_bus.call_blocking(self.device,path_batt_current,None,'GetValue', '', []) self.power = self.victron.session_bus.call_blocking(self.device,path_batt_power,None,'GetValue', '', []) self.temperature = self.victron.session_bus.call_blocking(self.device,path_batt_temperature,None,'GetValue', '', []) self.voltage_sum = self.victron.session_bus.call_blocking(self.device,path_batt_voltage_sum,None,'GetValue', '', []) self.voltage_lowest = self.victron.session_bus.call_blocking(self.device,path_batt_voltage_lowest,None,'GetValue', '', []) self.voltage_highest = self.victron.session_bus.call_blocking(self.device,path_batt_voltage_highest,None,'GetValue', '', []) self.voltage_diff = self.victron.session_bus.call_blocking(self.device,path_batt_voltage_diff,None,'GetValue', '', [])