232 lines
7.7 KiB
Python
232 lines
7.7 KiB
Python
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 Exception as e:
|
|
logger.info("Unable to connect to dbus {}: {!r}".format(path_setpoint, e))
|
|
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 Exception as e:
|
|
# keep the last known setpoint; None would break later comparisons
|
|
logger.error("Unable to set dbus {}: {!r}".format(path_setpoint, e))
|
|
|
|
|
|
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.name
|
|
|
|
|
|
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', '', [])
|
|
|