diff --git a/.pylintrc b/.pylintrc index 95c231c..7e859bd 100644 --- a/.pylintrc +++ b/.pylintrc @@ -71,7 +71,8 @@ disable=raw-checker-failed, wildcard-import, # to ignore messages like 'from helpers import *' unused-wildcard-import, # same as above logging-fstring-interpolation, # This diasbled the warning if usinf f-string in logging - fixme # TODO's wont be regonized. so this needs to be disabled + fixme, # TODO's wont be regonized. so this needs to be disabled + invalid-name # This is disabled to allow dbus-opendtu as a name. # Enable the message, report, category or checker with the given id(s). You can # either give multiple identifier separated by comma (,) or put this option # multiple time (only on the command line, not in the configuration file where @@ -516,5 +517,5 @@ preferred-modules= # Exceptions that will emit a warning when being caught. Defaults to # "BaseException, Exception". -overgeneral-exceptions=BaseException, - Exception +overgeneral-exceptions=builtins.BaseException, + builtins.Exception diff --git a/.vscode/settings.json b/.vscode/settings.json index 1b21aa5..76be030 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -8,4 +8,13 @@ "120" ], "editor.formatOnSave": true, + "python.testing.unittestArgs": [ + "-v", + "-s", + ".", + "-p", + "test_*.py" + ], + "python.testing.pytestEnabled": false, + "python.testing.unittestEnabled": true, } \ No newline at end of file diff --git a/README.md b/README.md index b5f00e8..f781904 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,9 @@ * [Template options](#template-options) * [Service names](#service-names) * [Videos how to install](#videos-how-to-install) + * [Use Cases](#use-cases) + * [Use Case 1: Using a Pv Inverter](#use-case-1-use-a-pv-inverter) + * [Use Case 2: Using a (Battery) Inverter](#use-case-2-use-a-battery-inverter) * [Usage](#usage) * [Check if script is running](#check-if-script-is-running) * [How to debug](#how-to-debug) @@ -121,11 +124,16 @@ This applies to each `TEMPLATE[X]` section. X is the number of Template starting | CUST_POLLING | Polling interval in ms for Device | | CUST_Total | Path in JSON 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_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_Default | [optional] Default value if no value is found in JSON | | CUST_Current | Path in JSON 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) *1| +| CUST_DCVoltage_Default | [optional] Default value if no value is found in JSON | | Phase | which Phase L1, L2, L3 to show| | 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) | @@ -141,30 +149,108 @@ Example for JSON PATH: use keywords separated by / The following servicenames are supported: * com.victronenergy.pvinverter -* com.victronenergy.inverter +* com.victronenergy.inverter (non-PV - see below) +* others might work but are not tested or undocumented yet -The difference between the two is that the first one is used as a PV inverter connected to the grid like a Fronius or SMA inverter. The second one is used for a battery inverter like a Victron AC Inverter. -For more Information about non-pv-inverters, see this [Issue #42](https://github.com/henne49/dbus-opendtu/issues/42). +**Note: Non-PV inverters are BETA! The functionality will be limited** (due to limited understanding of Victrons/Venus's behavior). -It is possible that other servicenames are supported, but not tested. If you have a device with a different servicename, please open an issue. +The difference between the two is that the first one (com.victronenergy.pvinverter) is used as a PV inverter connected to PV and the grid (like a Fronius or SMA inverter). +The second one (com.victronenergy.inverter) is used for a battery inverter like a Victron AC inverter and is - from Victron's view - not connected to the grid. +For more Information about non-PV inverters, see this [Issue #42](https://github.com/henne49/dbus-opendtu/issues/42). +Also, please note the use case about non-PV inverters below. + +It is possible that other servicenames are supported, but they have not been tested by us. If you have a device with a different servicename, please open an issue. Any help or research is welcome and appreciated. ### Videos how to install Here are some videos on how to install and use the script. They are in German, but you can use subtitles and auto-translate to your language. -*(Don't be confused that the config they used is not the actual one.)* +*(Don't be confused that the config they used is not the up-to-date.)* * Meine Energiewende * Schatten PV +### Use Cases + +In this section we describe some use cases and how to configure the script for them. + +#### **Use case 1: Use a PV inverter** + +In order to use a PV inverter, you need to know the IP address of the DTU (in my case Ahoy) and the servicename of the PV-Inverter. The servicename is `com.victronenergy.pvinverter`. + +A Basic configuration could look like this: + +```ini +[DEFAULT] +# Which DTU to be used ahoy, opendtu, template +DTU=ahoy + +#Possible Options for Log Level: CRITICAL, ERROR, WARNING, INFO, DEBUG, NOTSET +#To keep current.log small use ERROR +Logging=ERROR + +#IP of Device to query <-- THIS IS THE IP OF THE DTU +Host=192.168.1.74 + +### Ahoy Inverter +# AcPosition 0=AC input 1; 1=AC output; 2=AC output 2 +# 1st inverter +[INVERTER0] +Phase=L1 +DeviceInstance=34 +AcPosition=0 +``` + +The result will be that the first inverter is shown in the Remote Console of Venus OS. + +![Remote Console](./img/ahoy-as-pv-inverter.png) + +#### **Use case 2: Use a battery inverter** + +**NOTE: BETA - Victron never intended to use a Non-PV inverter (besides Multiplus, Quattro, etc.) to be connected to the existing grid directly (Grid synchronization).** + +In order to use a battery inverter, you need to know the IP address of the DTU (in my case Ahoy) and the servicename of the battery inverter. The servicename is `com.victronenergy.inverter`. + +The term battery inverter is used for a device that is connected to the grid and can discharge a battery. This is different from a PV inverter, which is only connected to PV-Modules and feeds in energy. + +You might want to use a battery inverter to use a battery to store energy from an MPPT charger / AC charger etc. and use it later. + +A Basic configuration could look like this: + +```ini +[DEFAULT] +# Which DTU to be used ahoy, opendtu, template +DTU=ahoy + +#Possible Options for Log Level: CRITICAL, ERROR, WARNING, INFO, DEBUG, NOTSET +#To keep current.log small use ERROR +Logging=ERROR + +#IP of Device to query <-- THIS IS THE IP OF THE DTU +Host=192.168.1.74 + +### Ahoy Inverter +# AcPosition 0=AC input 1; 1=AC output; 2=AC output 2 +# 1st inverter +[INVERTER0] +Phase=L1 +DeviceInstance=34 +AcPosition=0 +Servicename=com.victronenergy.inverter +``` + +The Result looks like this: + +![Battery-Inverter](./img/ahoy-as-inverter.png) + --- ## Usage -This are some useful commands which helps to use the script or to debug. +These are some useful commands which help to use the script or to debug. -### Check if script is running +### Check if the script is running -`svstat /service/dbus-opendtu` show if the service (our script) is running. If number of seconds show is low, the it is probably restarting and you should look into `/data/dbus-opendtu/current.log`. +`svstat /service/dbus-opendtu` show if the service (our script) is running. If the number of seconds shown is low, it is probably restarting and you should look into `/data/dbus-opendtu/current.log`. ### How to debug diff --git a/config.ini b/config.ini index 2ca37f7..ddf9b51 100644 --- a/config.ini +++ b/config.ini @@ -20,7 +20,7 @@ Logging=ERROR # if ts_last_success is older than this number of seconds, it is not used. # Set this to < 0 to disable this check. -MagAgeTsLastSuccess=600 +MaxAgeTsLastSuccess=600 # if this is not 0, then no values are actually sent via dbus to vrm/venus. DryRun=0 diff --git a/dbus-opendtu.py b/dbus-opendtu.py index 0c34228..e58a7a4 100644 --- a/dbus-opendtu.py +++ b/dbus-opendtu.py @@ -30,7 +30,7 @@ def main(): # configure logging config = configparser.ConfigParser() config.read(f"{(os.path.dirname(os.path.realpath(__file__)))}/config.ini") - logging_level = config["DEFAULT"]["Logging"] + logging_level = config["DEFAULT"]["Logging"].upper() dtuvariant = config["DEFAULT"]["DTU"] try: @@ -58,11 +58,19 @@ def main(): # Have a mainloop, so we can send/receive asynchronous calls to and from dbus DBusGMainLoop(set_as_default=True) - # formatting - def _kwh(p, v): return (str(round(v, 2)) + "KWh") - def _a(p, v): return (str(round(v, 1)) + "A") - def _w(p, v): return (str(round(v, 1)) + "W") - def _v(p, v): return (str(round(v, 1)) + "V") + # region formatting + def _kwh(_p, value: float) -> str: + return f"{round(value, 2)}KWh" + + def _a(_p, value: float) -> str: + return f"{round(value, 1)}A" + + def _w(_p, value: float) -> str: + return f"{round(value, 1)}W" + + def _v(_p, value: float) -> str: + return f"{round(value, 1)}V" + # endregion paths = { "/Ac/Energy/Forward": { diff --git a/dbus_service.py b/dbus_service.py index f842c31..393d822 100644 --- a/dbus_service.py +++ b/dbus_service.py @@ -1,7 +1,7 @@ '''DbusService and PvInverterRegistry''' # File specific rules -# pylint: disable=E0401,C0411,C0413,broad-except +# pylint: disable=broad-except, import-error, wrong-import-order, wrong-import-position # system imports: import configparser @@ -170,6 +170,18 @@ def _get_config(): config.read(f"{(os.path.dirname(os.path.realpath(__file__)))}/config.ini") return config + @staticmethod + def get_processed_meter_value(meter_data: dict, value: str, 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) + if isinstance(raw_value, (float, int)): + value = float(raw_value * float(factor)) + else: + value = default_value + + return value + # read config file def _read_config_dtu(self, actual_inverter): config = self._get_config() @@ -182,23 +194,23 @@ def _read_config_dtu(self, actual_inverter): {constants.DTUVARIANT_TEMPLATE}") self.deviceinstance = int(config[f"INVERTER{self.pvinverternumber}"]["DeviceInstance"]) self.acposition = int(get_config_value(config, "AcPosition", "INVERTER", self.pvinverternumber)) - self.signofliveinterval = config["DEFAULT"]["SignOfLifeLog"] - self.useyieldday = int(config["DEFAULT"]["useYieldDay"]) + self.signofliveinterval = get_config_value(config, "SignOfLifeLog", "DEFAULT", "", 1) + self.useyieldday = int(get_config_value(config, "useYieldDay", "DEFAULT", "", 0)) self.pvinverterphase = str(config[f"INVERTER{self.pvinverternumber}"]["Phase"]) self.host = get_config_value(config, "Host", "INVERTER", self.pvinverternumber) - self.username = get_config_value(config, "Username", "INVERTER", self.pvinverternumber) - self.password = get_config_value(config, "Password", "INVERTER", self.pvinverternumber) + self.username = get_config_value(config, "Username", "DEFAULT", "", self.pvinverternumber) + self.password = get_config_value(config, "Password", "DEFAULT", "", self.pvinverternumber) self.digestauth = is_true(get_config_value(config, "DigestAuth", "INVERTER", self.pvinverternumber, False)) try: - self.max_age_ts = int(config["DEFAULT"]["MagAgeTsLastSuccess"]) - except ValueError as ex: - logging.debug("MagAgeTsLastSuccess: %s", ex) - logging.debug("MagAgeTsLastSuccess not set, using default") + self.max_age_ts = int(config["DEFAULT"]["MaxAgeTsLastSuccess"]) + except (KeyError, ValueError) as ex: + logging.debug("MaxAgeTsLastSuccess: %s", ex) + logging.debug("MaxAgeTsLastSuccess not set, using default") self.max_age_ts = 600 self.dry_run = is_true(get_default_config(config, "DryRun", False)) - self.pollinginterval = int(config["DEFAULT"]["ESP8266PollingIntervall"]) + self.pollinginterval = int(get_config_value(config, "ESP8266PollingIntervall", "DEFAULT", "", 10000)) self.meter_data = 0 self.httptimeout = get_default_config(config, "HTTPTimeout", 2.5) @@ -207,9 +219,12 @@ def _read_config_template(self, template_number): self.pvinverternumber = template_number self.custpower = config[f"TEMPLATE{template_number}"]["CUST_Power"].split("/") self.custpower_factor = config[f"TEMPLATE{template_number}"]["CUST_Power_Mult"] + self.custpower_default = get_config_value(config, "CUST_Power_Default", "TEMPLATE", template_number, None) self.custtotal = config[f"TEMPLATE{template_number}"]["CUST_Total"].split("/") self.custtotal_factor = config[f"TEMPLATE{template_number}"]["CUST_Total_Mult"] + self.custtotal_default = get_config_value(config, "CUST_Total_Default", "TEMPLATE", template_number, None) self.custvoltage = config[f"TEMPLATE{template_number}"]["CUST_Voltage"].split("/") + self.custvoltage_default = get_config_value(config, "CUST_Voltage_Default", "TEMPLATE", template_number, None) self.custapipath = config[f"TEMPLATE{template_number}"]["CUST_API_PATH"] self.serial = str(config[f"TEMPLATE{template_number}"]["CUST_SN"]) self.pollinginterval = int(config[f"TEMPLATE{template_number}"]["CUST_POLLING"]) @@ -220,8 +235,8 @@ def _read_config_template(self, template_number): self.deviceinstance = int(config[f"TEMPLATE{template_number}"]["DeviceInstance"]) self.customname = config[f"TEMPLATE{template_number}"]["Name"] self.acposition = int(config[f"TEMPLATE{template_number}"]["AcPosition"]) - self.signofliveinterval = config["DEFAULT"]["SignOfLifeLog"] - self.useyieldday = int(config["DEFAULT"]["useYieldDay"]) + self.signofliveinterval = get_config_value(config, "SignOfLifeLog", "DEFAULT", "", 1) + self.useyieldday = int(get_config_value(config, "useYieldDay", "DEFAULT", "", 0)) self.pvinverterphase = str(config[f"TEMPLATE{template_number}"]["Phase"]) self.digestauth = is_true(get_config_value(config, "DigestAuth", "TEMPLATE", template_number, False)) @@ -231,6 +246,7 @@ def _read_config_template(self, template_number): # set to undefined because get_nested will solve this to 0 self.custcurrent = "[undefined]" logging.debug("CUST_Current not set") + self.custcurrent_default = get_config_value(config, "CUST_Current_Default", "TEMPLATE", template_number, None) try: self.custdcvoltage = config[f"TEMPLATE{template_number}"]["CUST_DCVoltage"].split("/") @@ -238,12 +254,14 @@ def _read_config_template(self, template_number): # set to undefined because get_nested will solve this to 0 self.custdcvoltage = "[undefined]" logging.debug("CUST_DCVoltage not set") + self.custdcvoltage_default = get_config_value( + config, "CUST_DCVoltage_Default", "TEMPLATE", template_number, None) try: - self.max_age_ts = int(config["DEFAULT"]["MagAgeTsLastSuccess"]) - except ValueError as ex: - logging.debug("MagAgeTsLastSuccess: %s", ex) - logging.debug("MagAgeTsLastSuccess not set, using default") + self.max_age_ts = int(config["DEFAULT"]["MaxAgeTsLastSuccess"]) + except (KeyError, ValueError) as ex: + logging.debug("MaxAgeTsLastSuccess: %s", ex) + logging.debug("MaxAgeTsLastSuccess not set, using default") self.max_age_ts = 600 self.dry_run = is_true(get_default_config(config, "DryRun", False)) @@ -597,12 +615,12 @@ def get_values_for_inverter(self): else 0) elif self.dtuvariant == constants.DTUVARIANT_TEMPLATE: - # logging.debug("JSON data: %s" % meter_data) - power = float(get_nested(meter_data, self.custpower) * float(self.custpower_factor)) - pvyield = float(get_nested(meter_data, self.custtotal) * float(self.custtotal_factor)) - voltage = float(get_nested(meter_data, self.custvoltage)) - dc_voltage = float(get_nested(meter_data, self.custdcvoltage)) - current = float(get_nested(meter_data, self.custcurrent)) + power = self.get_processed_meter_value( + 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) return (power, pvyield, current, voltage, dc_voltage) diff --git a/helpers.py b/helpers.py index f20b4c3..9028941 100644 --- a/helpers.py +++ b/helpers.py @@ -11,12 +11,12 @@ import logging -def get_config_value(config, name, inverter_or_template, pvinverternumber, defaultvalue=None): +def get_config_value(config, name, inverter_or_template, inverter_or_tpl_number, defaultvalue=None): '''check if config value exist in current inverter/template's section, otherwise throw error''' - if name in config[f"{inverter_or_template}{pvinverternumber}"]: - return config[f"{inverter_or_template}{pvinverternumber}"][name] + if name in config[f"{inverter_or_template}{inverter_or_tpl_number}"]: + return config[f"{inverter_or_template}{inverter_or_tpl_number}"][name] - if defaultvalue is None: + if defaultvalue is None and inverter_or_template == "INVERTER": raise ValueError(f"config entry '{name}' not found. " f"(Hint: Deprecated Host ONPREMISE entries must be moved to DEFAULT section.)") @@ -31,7 +31,7 @@ def get_default_config(config, name, defaultvalue): return defaultvalue -def get_nested(meter_data, path): +def get_value_by_path(meter_data, 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: @@ -45,6 +45,21 @@ def get_nested(meter_data, path): return value +def convert_to_expected_type(value: str, expected_type: [str, int, float, bool], + default: [None, str, int, float, bool]) -> [None, str, int, float, bool]: + ''' Try to convert value to expected_type, otherwise return default''' + try: + conversion_functions = { + str: str, + int: int, + float: float, + bool: is_true + } + return conversion_functions[expected_type](value) + except (ValueError, TypeError, KeyError): + return default + + def get_ahoy_field_by_name(meter_data, actual_inverter, fieldname, use_ch0_fld_names=True): '''get the value by name instead of list index''' # fetch value from record call: @@ -78,7 +93,7 @@ def get_ahoy_field_by_name(meter_data, actual_inverter, fieldname, use_ch0_fld_n def is_true(val): '''helper function to test for different true values''' - return val in (1, '1', True, "True", "true") + return val in (1, '1', True, "True", "TRUE", "true") def timeit(func): diff --git a/img/ahoy-as-inverter.png b/img/ahoy-as-inverter.png new file mode 100644 index 0000000..32e9c58 Binary files /dev/null and b/img/ahoy-as-inverter.png differ diff --git a/img/ahoy-as-pv-inverter.png b/img/ahoy-as-pv-inverter.png new file mode 100644 index 0000000..9ad4ebc Binary files /dev/null and b/img/ahoy-as-pv-inverter.png differ diff --git a/test_helpers.py b/test_helpers.py new file mode 100644 index 0000000..2b3d984 --- /dev/null +++ b/test_helpers.py @@ -0,0 +1,192 @@ +''' This file contains the unit tests for the helper functions in the helpers.py file. ''' + +# file ignores +# pylint: disable=too-many-instance-attributes + +import sys +import os +import unittest +from unittest.mock import MagicMock +import json +from helpers import * +sys.modules['vedbus'] = MagicMock() +sys.modules['dbus'] = MagicMock() +sys.modules['gi.repository'] = MagicMock() +sys.modules['requests'] = MagicMock() +sys.modules['requests.auth'] = MagicMock() + +import dbus_service # noqa pylint: disable=wrong-import-position + +# region Helper functions + + +def get_ahoy_meterdata(filename): + ''' Load the meter data from the json file. ''' + with open(filename, encoding="utf-8") as file_json: + json_meter_data = json.load(file_json) + json_meter_data["inverter"] = [] + for inverter_number in range(len(json_meter_data["iv"])): + if is_true(json_meter_data["iv"][inverter_number]): + iv_data = fetch_ahoy_iv_data(inverter_number) + while len(json_meter_data["inverter"]) < inverter_number: + # there was a gap + json_meter_data.append({}) + json_meter_data["inverter"].append(iv_data) + + return json_meter_data + + +def fetch_ahoy_iv_data(inverter_number): + ''' Load the inverter data from the json file. ''' + filename = f"./docs/ahoy_0.5.93_inverter-id-{inverter_number}.json" + # Check if the file exists, otherwise return an empty dict. + if not os.path.isfile(filename): + return {} + with open(filename, encoding="utf-8") as file_json: + data = json.load(file_json) + return data + + +# Load the meter data from the json file. +meter_data_ahoy = get_ahoy_meterdata(filename='./docs/ahoy_0.5.93_live.json') + +meter_data = json.loads( + '{"StatusSNS": {"Time": "2021-02-03T15:12:52", "Switch1": "ON", "ENERGY": ' + '{"TotalStartTime": "2020-01-05T12:41:22", "Total": 13.48712, "Yesterday": 0, ' + '"Today": 0, "Power": 190, "ApparentPower": 0, "ReactivePower": 0, "Factor": 0, ' + '"Voltage": 0, "Current": 0}}}') + +meter_data_null = json.loads( + '{"StatusSNS": {"Time": "2021-02-03T15:12:52", "Switch1": "ON", "ENERGY": ' + '{"TotalStartTime": "2020-01-05T12:41:22", "Total": 13.48712, "Yesterday": 0, ' + '"Today": 0, "Power": null, "ApparentPower": null, "ReactivePower": null, "Factor": null, ' + '"Voltage": 225.66, "Current": null}}}') +# endregion + + +class TestHelpersFunctions(unittest.TestCase): + ''' This class contains the unit tests for the helper functions in the helpers.py file. ''' + + def setUp(self): + ''' Setup the test environment. ''' + # TODO: Create a mock config file and use that instead of the real one. + self.config = dbus_service.DbusService._get_config() # pylint: disable=protected-access + + self.custpower = self.config["TEMPLATE0"]["CUST_Power"].split("/") + self.custpower_factor = self.config["TEMPLATE0"]["CUST_Power_Mult"] + self.custpower_default = get_config_value(self.config, "CUST_Power_Default", "TEMPLATE", 0, None) + self.custtotal = self.config["TEMPLATE0"]["CUST_Total"].split("/") + self.custtotal_factor = self.config["TEMPLATE0"]["CUST_Total_Mult"] + self.custtotal_default = get_config_value(self.config, "CUST_Total_Default", "TEMPLATE", 0, None) + self.custvoltage = self.config["TEMPLATE0"]["CUST_Voltage"].split("/") + self.custvoltage_default = get_config_value( + self.config, "CUST_Voltage_Default", "TEMPLATE", 0, None) + self.custcurrent = self.config["TEMPLATE0"]["CUST_Current"].split("/") + self.custcurrent_default = get_config_value( + self.config, "CUST_Current_Default", "TEMPLATE", 0) + + def test_get_config_value(self): + ''' Test the get_config_value() function. ''' + self.assertEqual(get_config_value(self.config, "Phase", "INVERTER", 0), "L1") + self.assertEqual(get_config_value(self.config, "Username", "TEMPLATE", 0), "") + self.assertEqual(get_config_value(self.config, "not_exist", "TEMPLATE", 0, "default"), "default") + with self.assertRaises(ValueError): + get_config_value(self.config, "not_exist", "INVERTER", 0) + + def test_get_default_config(self): + ''' Test the get_default_config() function. ''' + self.assertEqual(get_default_config(self.config, "Phase", "L1"), "L1") + self.assertEqual(get_default_config(self.config, "not_exist", "default"), "default") + self.assertEqual(get_default_config(self.config, "DTU", "empty"), "opendtu") + + def test_get_value_by_path(self): + ''' Test the get_nested() function. ''' + self.assertEqual(get_value_by_path(meter_data, self.custpower), 190) + self.assertEqual(get_value_by_path(meter_data, self.custtotal), 13.48712) + self.assertEqual(get_value_by_path(meter_data, ["StatusSNS", "ENERGY", "not_there"]), 0) + self.assertEqual(get_value_by_path(meter_data, ["StatusSNS", "Switch1"]), "ON") + + def test_convert_to_expected_type(self): + ''' Test the convert_to_expected_type() function. ''' + self.assertEqual(convert_to_expected_type("test", str, "default"), "test") + self.assertEqual(convert_to_expected_type("test", str, None), "test") + self.assertEqual(convert_to_expected_type("test", int, 0), 0) + self.assertEqual(convert_to_expected_type("test", int, None), None) + self.assertEqual(convert_to_expected_type("test", float, 0.0), 0.0) + self.assertEqual(convert_to_expected_type("test", float, None), None) + self.assertEqual(convert_to_expected_type("test", bool, False), False) + self.assertEqual(convert_to_expected_type("1", bool, None), True) + self.assertEqual(convert_to_expected_type(None, None, None), None) + + def test_get_ahoy_field_by_name(self): + ''' Test the get_ahoy_field_by_name() function. ''' + self.assertEqual(get_ahoy_field_by_name(meter_data_ahoy, 0, "P_AC"), 223.7) + self.assertEqual(get_ahoy_field_by_name(meter_data_ahoy, 0, "YieldDay"), 2223) + self.assertEqual(get_ahoy_field_by_name(meter_data_ahoy, 0, "YieldTotal"), 422.603) + self.assertEqual(get_ahoy_field_by_name(meter_data_ahoy, 0, "U_AC"), 229.5) + self.assertEqual(get_ahoy_field_by_name(meter_data_ahoy, 0, "U_DC", False), 33.3) + self.assertEqual(get_ahoy_field_by_name(meter_data_ahoy, 0, "I_AC"), 0.98) + self.assertEqual(get_ahoy_field_by_name(meter_data_ahoy, 0, "I_DC", False), 1.75) + self.assertEqual(get_ahoy_field_by_name(meter_data_ahoy, 0, "P_DC", False), 58.1) + + def test_is_true(self): + ''' Test the is_true() function. ''' + self.assertEqual(is_true("1"), True) + self.assertEqual(is_true("true"), True) + self.assertEqual(is_true("True"), True) + self.assertEqual(is_true("TRUE"), True) + self.assertEqual(is_true("0"), False) + self.assertEqual(is_true("false"), False) + self.assertEqual(is_true("False"), False) + self.assertEqual(is_true("FALSE"), False) + self.assertEqual(is_true("test"), False) + self.assertEqual(is_true(""), False) + self.assertEqual(is_true(None), False) + + def test_timeit(self): + ''' Test the timeit() function. ''' + @timeit + def test_function(): + ''' Test function. ''' + return 1 + + self.assertEqual(test_function(), 1) + + def test_part_get_values_for_inverts(self): + ''' Test part of get_values_for_inverter() function, which is in dbus_service + but heavily uses functions in helpers.py. ''' + + power = dbus_service.DbusService.get_processed_meter_value( + meter_data_null, + self.custpower, + self.custpower_default, + self.custpower_factor + ) + + pvyield = dbus_service.DbusService.get_processed_meter_value( + meter_data_null, + self.custtotal, + self.custtotal_default, + self.custtotal_factor + ) + + voltage = dbus_service.DbusService.get_processed_meter_value( + meter_data_null, + self.custvoltage, + self.custpower_default, + ) + + current = dbus_service.DbusService.get_processed_meter_value( + meter_data_null, + self.custcurrent, + self.custpower_default, + ) + + self.assertEqual(power, None) + self.assertEqual(pvyield, 13.48712) + self.assertEqual(voltage, 225.66) + self.assertEqual(current, None) + + +if __name__ == '__main__': + unittest.main()