Testcases for template

README.md

@@ -127,29 +127,30 @@ This applies to each `TEMPLATE[X]` section. X is the number of Template starting
 | CUST_SN | Serialnumber to register device in VenusOS|
 | CUST_API_PATH | Location of REST API Path for JSON to be used |
 | CUST_POLLING | Polling interval in ms for Device |
-| CUST_Total | Path in JSON where to find total Energy |
+| CUST_Total | Path in JSON *4 where to find total Energy |
 | CUST_Total_Mult | Multiplier to convert W per minute for example in kWh|
 | CUST_Total_Default | [optional] Default value if no value is found in JSON |
-| CUST_Power | Path in JSON where to find actual Power |
+| CUST_Power | Path in JSON *4 where to find actual Power |
 | CUST_Power_Mult | Multiplier to convert W in negative or positive |
 | CUST_Power_Default | [optional] Default value if no value is found in JSON |
-| CUST_Voltage | Path in JSON where to find actual Voltage |
+| CUST_Voltage | Path in JSON *4 where to find actual Voltage |
 | CUST_Voltage_Default | [optional] Default value if no value is found in JSON |
-| CUST_Current | Path in JSON where to find actual Current |
+| CUST_Current | Path in JSON *4 where to find actual Current |
 | CUST_Current_Default | [optional] Default value if no value is found in JSON |
-| CUST_DCVoltage | Path in JSON where to find actual DC Voltage (e.g. Batterie voltage) *2|
+| CUST_DCVoltage | Path in JSON *4 where to find actual DC Voltage (e.g. Batterie voltage) *2|
 | CUST_DCVoltage_Default | [optional] Default value if no value is found in JSON |
 | Phase | which Phase L1, L2, L3 to show; use 3P for three-phase-inverters *3 |
 | DeviceInstance | Unique ID identifying the OpenDTU in Venus OS|
 | AcPosition | Position shown in Remote Console (0=AC input 1; 1=AC output; 2=AC input 2 please do not use) |
 | Name | Name to be shown in VenusOS, use a descriptive name |
 | Servicename | e.g. com.victronenergy.pvinverter see [Service names](#service-names) |
 
-Example for JSON PATH: use keywords separated by /
-
 *2: is only used if Servicename is com.victronenergy.inverter
+
 *3: Use 3P to split power equally over three phases (use this for Hoymiles three-phase micro-inverters as they report total power only, not seperated by phase).
 
+*4: Path in JSON: use keywords and array index numbers separated by `/`. Example (compare [tasmota_shelly_2pm.json](docs/tasmota_shelly_2pm.json)): `StatusSNS/ENERGY/Current/0` fetches dictionary (map) entry `StatusSNS` containting an entry `ENERGY` containing an entry `Current` containing an array where the first element (index 0) is taken.
+
 ### Service names
 
 The following servicenames are supported:

dbus-opendtu.py

@@ -101,6 +101,7 @@ def _v(_p, value: float) -> str:
             "/Ac/L3/Energy/Forward": {"initial": None, "textformat": _kwh},
             "/Ac/Out/L1/I": {"initial": None, "textformat": _a},
             "/Ac/Out/L1/V": {"initial": None, "textformat": _v},
+            "/Ac/Out/L1/P": {"initial": None, "textformat": _w},
             "/Dc/0/Voltage": {"initial": None, "textformat": _v},
         }
 
@@ -113,7 +114,7 @@ def _v(_p, value: float) -> str:
                 actual_inverter=0,
             )
 
-            if number_of_inverters == 0: 
+            if number_of_inverters == 0:
                 number_of_inverters = service.get_number_of_inverters()
 
             if number_of_inverters > 1:

dbus_service.py

@@ -143,8 +143,8 @@ def __init__(
                 onchangecallback=self._handlechangedvalue,
             )
 
-        # add _sign_of_life 'timer' to get feedback in log every 5minutes
         gobject.timeout_add(self._get_sign_of_life_interval() * 60 * 1000, self._sign_of_life)
+        gobject.timeout_add(self._get_polling_interval(), self._update)
 
     @staticmethod
     def get_ac_inverter_state(current):
@@ -171,15 +171,18 @@ def _get_config():
         return config
 
     @staticmethod
-    def get_processed_meter_value(meter_data: dict, value: str, default_value: any, factor: int = 1) -> any:
+    def get_processed_meter_value(meter_data: dict, path_to_value, default_value: any, factor: int = 1) -> any:
         '''return the processed meter value by applying the factor and return a default value due an Exception'''
-        get_raw_value = get_value_by_path(meter_data, value)
-        raw_value = convert_to_expected_type(get_raw_value, float, default_value)
+        raw_value = get_value_by_path(meter_data, path_to_value)
+        logging.debug(f"get_processed_meter_value: path_to_value={path_to_value}, raw_value={raw_value}")
+        raw_value = convert_to_expected_type(raw_value, float, default_value)
         if isinstance(raw_value, (float, int)):
             value = float(raw_value * float(factor))
         else:
             value = default_value
 
+        logging.debug(f"get_processed_meter_value(..., path_to_value={path_to_value}, default_value={default_value}, factor={factor})"
+                      f" returns {value}")
         return value
 
     # read config file
@@ -287,8 +290,6 @@ def _get_serial(self, pvinverternumber):
         elif self.dtuvariant == constants.DTUVARIANT_TEMPLATE:
             serial = self.serial
 
-        gobject.timeout_add(self._get_polling_interval(), self._update)
-
         return serial
 
     def _get_name(self):
@@ -486,7 +487,7 @@ def fetch_url(self, url, try_number=1):
             else:
                 raise
 
-    def _get_data(self):
+    def _get_data(self) -> dict:
         if self._test_meter_data:
             return self._test_meter_data
         if not DbusService._meter_data:
@@ -534,6 +535,7 @@ def _sign_of_life(self):
         return True
 
     def _update(self):
+        logging.debug("_update")
         successful = False
         try:
             # update data from DTU once per _update call:
@@ -603,13 +605,13 @@ def get_values_for_inverter(self):
             # OpenDTU v24.2.12 breaking API changes 2024-02-19
             if "AC" in meter_data["inverters"][self.pvinverternumber]:
                 root_meter_data = meter_data["inverters"][self.pvinverternumber]
-                firmware_v24_2_12_or_newer=True
+                firmware_v24_2_12_or_newer = True
             else:
                 inverter_serial = meter_data["inverters"][self.pvinverternumber]["serial"]
                 logging.info(f"Inverter #{self.pvinverternumber} Serial: {inverter_serial}")
                 root_meter_data = self.fetch_opendtu_inverter_data(inverter_serial)["inverters"][0]
                 logging.debug(f"{root_meter_data}")
-                firmware_v24_2_12_or_newer=False
+                firmware_v24_2_12_or_newer = False
 
             producing = is_true(root_meter_data["producing"])
             power = (root_meter_data["AC"]["0"]["Power"]["v"]
@@ -631,31 +633,48 @@ def get_values_for_inverter(self):
                 meter_data, self.custpower, self.custpower_default, self.custpower_factor)
             pvyield = self.get_processed_meter_value(
                 meter_data, self.custtotal, self.custtotal_default, self.custtotal_factor)
-            voltage = self.get_processed_meter_value(meter_data, self.custvoltage, self.custpower_default)
-            current = self.get_processed_meter_value(meter_data, self.custcurrent, self.custpower_default)
+            voltage = self.get_processed_meter_value(meter_data, self.custvoltage, self.custdcvoltage_default)
+            current = self.get_processed_meter_value(meter_data, self.custcurrent, self.custcurrent_default)
 
         return (power, pvyield, current, voltage, dc_voltage)
 
     def set_dbus_values(self):
         '''read data and set dbus values'''
         (power, pvyield, current, voltage, dc_voltage) = self.get_values_for_inverter()
+        state = self.get_ac_inverter_state(current)
 
         # This will be refactored later in classes
         if self._servicename == "com.victronenergy.inverter":
+            # see https://github.com/victronenergy/venus/wiki/dbus#inverter
             self._dbusservice["/Ac/Out/L1/V"] = voltage
             self._dbusservice["/Ac/Out/L1/I"] = current
+            self._dbusservice["/Ac/Out/L1/P"] = power
             self._dbusservice["/Dc/0/Voltage"] = dc_voltage
-            self._dbusservice["/State"] = self.get_ac_inverter_state(current)
+            self._dbusservice["/Ac/Power"] = power
+
+            self._dbusservice["/Ac/Energy/Forward"] = pvyield
+            self._dbusservice["/State"] = state
+            self._dbusservice["/Mode"] = 2  # Switch position: 2=Inverter on; 4=Off; 5=Low Power/ECO
+
+            self._dbusservice["/Ac/L1/Current"] = current
+            self._dbusservice["/Ac/L1/Energy/Forward"] = pvyield
+            self._dbusservice["/Ac/L1/Power"] = power
+            self._dbusservice["/Ac/L1/Voltage"] = voltage
 
             logging.debug(f"Inverter #{self.pvinverternumber} Voltage (/Ac/Out/L1/V): {voltage}")
             logging.debug(f"Inverter #{self.pvinverternumber} Current (/Ac/Out/L1/I): {current}")
+
+            logging.debug(f"Inverter #{self.pvinverternumber} Current (/Dc/0/Voltage): {dc_voltage}")
+            logging.debug(f"Inverter #{self.pvinverternumber} Voltage (/Ac/Power): {power}")
+            logging.debug(f"Inverter #{self.pvinverternumber} Current (/Ac/Energy/Forward): {pvyield}")
+            logging.debug(f"Inverter #{self.pvinverternumber} Current (/State): {state}")
             logging.debug("---")
         else:
             # three-phase inverter: split total power equally over all three phases
             if ("3P" == self.pvinverterphase):
                 powerthird = power/3
 
-                #Single Phase Voltage = (3-Phase Voltage) / (sqrt(3))
+                # Single Phase Voltage = (3-Phase Voltage) / (sqrt(3))
                 # This formula assumes that the three-phase voltage is balanced and that
                 # the phase angles are 120 degrees apart
                 # sqrt(3) = 1.73205080757 <-- So we do not need to include Math Library

docs/tasmota_shelly_2pm.json

@@ -0,0 +1,39 @@
+{
+    "StatusSNS": {
+        "Time": "2023-09-24T12:13:12+02:00",
+        "Switch1": "OFF",
+        "Switch2": "ON",
+        "ANALOG": {
+            "Temperature": 54.1
+        },
+        "ENERGY": {
+            "TotalStartTime": "2023-06-06T13:21:33",
+            "Total": 78.85,
+            "Yesterday": 0.002,
+            "Today": 0.315,
+            "Power": [
+                160,
+                0
+            ],
+            "ApparentPower": [
+                169,
+                0
+            ],
+            "ReactivePower": [
+                33,
+                0
+            ],
+            "Factor": [
+                0.95,
+                0
+            ],
+            "Frequency": 50,
+            "Voltage": 235,
+            "Current": [
+                0.734,
+                0
+            ]
+        },
+        "TempUnit": "C"
+    }
+}

helpers.py

@@ -31,7 +31,7 @@ def get_default_config(config, name, defaultvalue):
     return defaultvalue
 
 
-def get_value_by_path(meter_data, path):
+def get_value_by_path(meter_data: dict, path):
     '''Try to extract 'path' from nested array 'meter_data' (derived from json document) and return the found value'''
     value = meter_data
     for path_entry in path:
@@ -88,6 +88,8 @@ def get_ahoy_field_by_name(meter_data, actual_inverter, fieldname, use_ch0_fld_n
         dc_channel_index = 1  # 1 = DC1, 2 = DC2 etc.
         data = meter_data["inverter"][actual_inverter]["ch"][dc_channel_index][data_index]
 
+    logging.debug(f"get_ahoy_field_by_name(..., actual_inverter={actual_inverter}, fieldname={fieldname}, {use_ch0_fld_names})"
+                  f" returns '{data}'")
     return data
 
 

tests.py

@@ -11,12 +11,14 @@
 
 # Victron imports:
 from dbus_service import DbusService
+from helpers import get_value_by_path
 
 
 OPENDTU_TEST_DATA_FILE = "docs/opendtu_status.json"
 AHOY_TEST_DATA_FILE_LIVE = "docs/ahoy_0.5.93_live.json"
 AHOY_TEST_DATA_FILE_RECORD = "docs/ahoy_0.5.93_record-live.json"
 AHOY_TEST_DATA_FILE_IV_0 = "docs/ahoy_0.5.93_inverter-id-0.json"
+TEMPLATE_TASMOTA_TEST_DATA_FILE = "docs/tasmota_shelly_2pm.json"
 
 
 def test_opendtu_reachable(test_service):
@@ -78,6 +80,12 @@ def load_ahoy_test_data():
     return test_data
 
 
+def load_template_tasmota_test_data():
+    '''Load Test data for Template for tasmota case'''
+    test_data = load_json_file(TEMPLATE_TASMOTA_TEST_DATA_FILE)
+    return test_data
+
+
 def test_ahoy_values(test_service):
     '''test with ahoy data'''
     test_service.set_dtu_variant(constants.DTUVARIANT_AHOY)
@@ -111,11 +119,49 @@ def test_ahoy_get_number_of_inverters(test_service):
     assert test_service.get_number_of_inverters() == 3
 
 
+def test_get_value_by_path():
+    test_meter_data = {
+        "a": 1,
+        "b": {
+            "c": 3,
+            "arr": ["x", "y"],
+        }
+    }
+    assert 1 == get_value_by_path(test_meter_data, ["a"])
+    assert 3 == get_value_by_path(test_meter_data, ["b", "c"])
+    assert "y" == get_value_by_path(test_meter_data, ["b", "arr", 1])  # not: ["b", "arr[1]"]
+
+
+def test_template_values(test_service):
+    '''test with template test data for tasmota'''
+    test_service.set_dtu_variant(constants.DTUVARIANT_TEMPLATE)
+    test_service.custpower = "StatusSNS/ENERGY/Power/0".split("/")
+    test_service.custcurrent = "StatusSNS/ENERGY/Current/0".split("/")
+    test_service.custpower_default = 999
+    test_service.custcurrent_default = 999
+    test_service.custpower_factor = 2
+    test_service.custtotal_default = 99
+    test_service.custtotal_factor = 1
+    test_service.custvoltage = "StatusSNS/ENERGY/Voltage".split("/")
+    test_service.custdcvoltage_default = 99.9
+    test_service.custtotal = "StatusSNS/ENERGY/Today".split("/")
+
+    test_data = load_template_tasmota_test_data()
+
+    test_service.set_test_data(test_data)
+    logging.debug("starting test for test_template_values")
+    (power, pvyield, current, voltage, dc_voltage) = test_service.get_values_for_inverter()
+    print(power, pvyield, current, voltage, dc_voltage)
+    assert (power, pvyield, current, voltage, dc_voltage) == (320.0, 0.315, 0.734, 235, None)
+
+
 def run_tests():
     '''function to run tests'''
+    test_get_value_by_path()
     test_service = DbusService(servicename="testing", paths="dummy", actual_inverter=0)
     test_opendtu_reachable(test_service)
     test_opendtu_producing(test_service)
     test_ahoy_values(test_service)
     test_ahoy_timestamp(test_service)
+    test_template_values(test_service)
     logging.debug("tests have passed")