From 7a52f14b11777b1a7b4c64596cf32c650667d713 Mon Sep 17 00:00:00 2001 From: CI Date: Thu, 24 Sep 2026 21:16:37 +0100 Subject: [PATCH 1/2] GE Cloud: parse option validation text with bracketed labels without crashing (#5217) Co-Authored-By: Claude Opus 5.5 (1M context) --- apps/predbat/gecloud.py | 38 ++++++--- apps/predbat/tests/test_ge_cloud.py | 117 ++++++++++++++++++++++++++++ 2 files changed, 145 insertions(+), 10 deletions(-) diff --git a/apps/predbat/gecloud.py b/apps/predbat/gecloud.py index dd63917ae..c72f6fd9a 100644 --- a/apps/predbat/gecloud.py +++ b/apps/predbat/gecloud.py @@ -363,6 +363,22 @@ def regname_to_ha(name): return name +VALIDATION_OPTIONS_PREFIX = "Value must be one of:" + + +def parse_validation_options(validation): + """ + Return the option labels from a 'Value must be one of: (a, b, c)' validation string, or None when it cannot be parsed + """ + if not isinstance(validation, str) or not validation.startswith(VALIDATION_OPTIONS_PREFIX) or "(" not in validation: + return None + # Split on the first '(' and drop only the final ')' so option labels that contain brackets survive intact + post = validation.split("(", 1)[1] + post = post.rsplit(")", 1)[0] + post = post.replace(", ", ",") + return post.split(",") + + def normalise_register_time(value): """ Return a register time value as an HH:MM string, or None when it is not a time @@ -730,11 +746,12 @@ async def select_event(self, entity_id, value): if validation_rule.startswith("in:"): options_values = validation_rule.split(":")[1].split(",") - if validation and validation.startswith("Value must be one of:"): - pre, post = validation.split("(") - post = post.replace(")", "") - post = post.replace(", ", ",") - options_text = post.split(",") + if validation and validation.startswith(VALIDATION_OPTIONS_PREFIX): + options_text = parse_validation_options(validation) + if options_text is None: + self.log("GECloud: Warn: Unable to parse options for setting {} {} from validation '{}'".format(device, key, validation)) + + if options_text is not None: if new_value not in options_text: self.log("GECloud: Error: Invalid option {} for setting {} {}, valid values are {}".format(new_value, device, key, options_text)) return @@ -1305,11 +1322,12 @@ async def publish_registers(self, device, registers, select_key=None): attributes["device_class"] = "power" attributes["unit_of_measurement"] = "W" - if validation and validation.startswith("Value must be one of:"): - pre, post = validation.split("(") - post = post.replace(")", "") - post = post.replace(", ", ",") - options_text = post.split(",") + if validation and validation.startswith(VALIDATION_OPTIONS_PREFIX): + validation_options = parse_validation_options(validation) + if validation_options is None: + self.log("GECloud: Warn: Unable to parse options for setting {} {} from validation '{}'".format(device, key, validation)) + else: + options_text = validation_options if is_select_time or is_select_options: entity_name = f"select.{self.prefix}_gecloud_{device}" diff --git a/apps/predbat/tests/test_ge_cloud.py b/apps/predbat/tests/test_ge_cloud.py index 1f0e8d854..1df369633 100644 --- a/apps/predbat/tests/test_ge_cloud.py +++ b/apps/predbat/tests/test_ge_cloud.py @@ -292,11 +292,14 @@ def test_ge_cloud(my_predbat=None): ("switch_event", _test_switch_event, "Switch event handler"), ("number_event", _test_number_event, "Number event handler"), ("select_event", _test_select_event, "Select event handler"), + ("select_event_bracketed_options", _test_select_event_bracketed_options, "Select event with bracketed or unparseable option validation"), + ("parse_validation_options", _test_parse_validation_options, "Parse options from validation text"), ("publish_status", _test_publish_status, "Publish status"), ("publish_meter", _test_publish_meter, "Publish meter"), ("publish_info", _test_publish_info, "Publish info"), ("publish_info_soh", _test_publish_info_soh, "Publish info SOH calculation"), ("publish_registers", _test_publish_registers, "Publish registers"), + ("publish_registers_bracketed_options", _test_publish_registers_bracketed_options, "Publish registers with bracketed or unparseable option validation"), ("publish_evc_data", _test_publish_evc_data, "Publish EVC data"), ("automatic_config", _test_async_automatic_config, "Automatic config"), ("discovery_direct", _test_build_discovery_battery_only_direct, "build_discovery: single battery, direct composition"), @@ -4000,6 +4003,77 @@ async def mock_publish(*args, **kwargs): return run_async(test()) +def _test_select_event_bracketed_options(my_predbat): + """Test select event accepts option labels that contain brackets and falls back to 'in:' rules on unparseable validation (GH#5217)""" + + async def test(): + ge_cloud = MockGECloudDirect() + entity_id = "select.predbat_gecloud_test123_ems_mode" + ge_cloud.register_entity_map[entity_id] = {"key": 90, "device": "test123"} + ge_cloud.settings["test123"] = {90: {"value": "Eco (Paused)", "validation": "Value must be one of: (Eco (Paused), Timed Demand, Export (AC3))", "validation_rules": []}} + + write_calls = [] + + async def mock_write(serial, setting_id, value): + write_calls.append(value) + return {"value": value} + + async def mock_publish(*args, **kwargs): + pass + + with patch("gecloud.asyncio.sleep", new_callable=AsyncMock): + ge_cloud.async_write_inverter_setting = mock_write + ge_cloud.publish_registers = mock_publish + + await ge_cloud.select_event(entity_id, "Export (AC3)") + if write_calls != ["Export (AC3)"]: + print("ERROR: Expected bracketed option to be written, got {}".format(write_calls)) + return 1 + + await ge_cloud.select_event(entity_id, "Bogus") + if write_calls != ["Export (AC3)"]: + print("ERROR: Expected invalid option to be rejected, got {}".format(write_calls)) + return 1 + + # Validation text without any bracket cannot be parsed: fall back to the 'in:' rule rather than crash + ge_cloud.settings["test123"][90] = {"value": "1", "validation": "Value must be one of:", "validation_rules": ["in:1,2"]} + await ge_cloud.select_event(entity_id, "2") + if write_calls != ["Export (AC3)", "2"]: + print("ERROR: Expected 'in:' fallback to accept '2', got {}".format(write_calls)) + return 1 + await ge_cloud.select_event(entity_id, "3") + if write_calls != ["Export (AC3)", "2"]: + print("ERROR: Expected 'in:' fallback to reject '3', got {}".format(write_calls)) + return 1 + if not any("Unable to parse options" in message for message in ge_cloud.log_messages): + print("ERROR: Expected a warning for unparseable validation, got {}".format(ge_cloud.log_messages)) + return 1 + + return 0 + + return run_async(test()) + + +def _test_parse_validation_options(my_predbat): + """Test parsing option labels out of GE Cloud 'Value must be one of:' validation strings (GH#5217)""" + from gecloud import parse_validation_options + + cases = [ + ("Value must be one of: (00:00, 00:30, 01:00)", ["00:00", "00:30", "01:00"]), + ("Value must be one of: (Eco (Paused), Timed Demand, Export (AC3))", ["Eco (Paused)", "Timed Demand", "Export (AC3)"]), + ("Value must be one of:", None), + ("Value must be between 0 and 100", None), + ("", None), + (None, None), + ] + for validation, expected in cases: + result = parse_validation_options(validation) + if result != expected: + print("ERROR: parse_validation_options({!r}) expected {}, got {}".format(validation, expected, result)) + return 1 + return 0 + + # ============================================================================= # Publishing Tests # ============================================================================= @@ -4205,6 +4279,49 @@ def _test_publish_registers(my_predbat): return 0 +def _test_publish_registers_bracketed_options(my_predbat): + """Test publishing registers whose validation text has bracketed labels or no brackets at all does not abort the device (GH#5217)""" + ge_cloud = MockGECloudDirect() + ge_cloud.config_args["prefix"] = "predbat" + + registers = { + 90: {"name": "EMS Mode", "validation_rules": ["in:0,1,2"], "validation": "Value must be one of: (Eco (Paused), Timed Demand, Export (AC3))", "value": "Timed Demand"}, + 91: {"name": "AC3 Mode", "validation_rules": ["in:0,1"], "validation": "Value must be one of:", "value": "0"}, + 92: {"name": "Enable AC Charge", "validation_rules": ["boolean"], "validation": "", "value": "1"}, + } + ge_cloud.register_list["test123"] = registers + + run_async(ge_cloud.publish_registers("test123", registers)) + + ems_entity = "select.predbat_gecloud_test123_ems_mode" + if ems_entity not in ge_cloud.dashboard_items: + print("ERROR: Expected {} to be published".format(ems_entity)) + return 1 + options = ge_cloud.dashboard_items[ems_entity]["attributes"]["options"] + if options != ["Eco (Paused)", "Timed Demand", "Export (AC3)"]: + print("ERROR: Expected bracketed labels to survive intact, got {}".format(options)) + return 1 + + ac3_entity = "select.predbat_gecloud_test123_ac3_mode" + if ac3_entity not in ge_cloud.dashboard_items: + print("ERROR: Expected {} to be published".format(ac3_entity)) + return 1 + options = ge_cloud.dashboard_items[ac3_entity]["attributes"]["options"] + if options != ["0", "1"]: + print("ERROR: Expected unparseable validation to fall back to 'in:' options, got {}".format(options)) + return 1 + if not any("Unable to parse options" in message for message in ge_cloud.log_messages): + print("ERROR: Expected a warning for unparseable validation, got {}".format(ge_cloud.log_messages)) + return 1 + + # Registers after the problem ones must still be published + if "switch.predbat_gecloud_test123_enable_ac_charge" not in ge_cloud.dashboard_items: + print("ERROR: Expected later registers to still be published") + return 1 + + return 0 + + def _test_publish_evc_data(my_predbat): """Test publishing EV charger data""" From 99aafa86373b5076f313cc415d5bc5bd43b04617 Mon Sep 17 00:00:00 2001 From: CI Date: Sat, 26 Sep 2026 11:13:29 +0100 Subject: [PATCH 2/2] fix(gecloud): address #5218 review - empty/comma-in-bracket options, warn once, non-string guard Co-Authored-By: Claude Opus 5.5 (1M context) --- apps/predbat/gecloud.py | 31 +++++++++++++++++++---- apps/predbat/tests/test_ge_cloud.py | 38 +++++++++++++++++++++++++++++ 2 files changed, 64 insertions(+), 5 deletions(-) diff --git a/apps/predbat/gecloud.py b/apps/predbat/gecloud.py index dfd52671e..195a851e9 100644 --- a/apps/predbat/gecloud.py +++ b/apps/predbat/gecloud.py @@ -402,8 +402,24 @@ def parse_validation_options(validation): # Split on the first '(' and drop only the final ')' so option labels that contain brackets survive intact post = validation.split("(", 1)[1] post = post.rsplit(")", 1)[0] - post = post.replace(", ", ",") - return post.split(",") + if not post.strip(): + return None + # Split only on commas outside brackets so a label such as 'Pause (Battery, Grid)' stays whole + options = [] + label = "" + depth = 0 + for char in post: + if char == "(": + depth += 1 + elif char == ")": + depth = max(depth - 1, 0) + elif char == "," and depth == 0: + options.append(label) + label = "" + continue + label += char + options.append(label) + return [option[1:] if index > 0 and option.startswith(" ") else option for index, option in enumerate(options)] def normalise_register_time(value): @@ -582,6 +598,8 @@ def initialize(self, ge_cloud_direct, api_key, automatic, automatic_evc=False, e # Battery inverters already reported as overriding the EMS, so a standing misconfiguration # raises the Predbat status once rather than on every settings refresh self.ems_slot_warned = set() + # (device, key) registers whose option validation text could not be parsed, so the warning is logged once + self.validation_parse_warned = set() self.devices_dict = {} self.device_list = [] self.ems_device = None @@ -776,7 +794,7 @@ async def select_event(self, entity_id, value): if validation_rule.startswith("in:"): options_values = validation_rule.split(":")[1].split(",") - if validation and validation.startswith(VALIDATION_OPTIONS_PREFIX): + if isinstance(validation, str) and validation.startswith(VALIDATION_OPTIONS_PREFIX): options_text = parse_validation_options(validation) if options_text is None: self.log("GECloud: Warn: Unable to parse options for setting {} {} from validation '{}'".format(device, key, validation)) @@ -1352,10 +1370,13 @@ async def publish_registers(self, device, registers, select_key=None): attributes["device_class"] = "power" attributes["unit_of_measurement"] = "W" - if validation and validation.startswith(VALIDATION_OPTIONS_PREFIX): + if isinstance(validation, str) and validation.startswith(VALIDATION_OPTIONS_PREFIX): validation_options = parse_validation_options(validation) if validation_options is None: - self.log("GECloud: Warn: Unable to parse options for setting {} {} from validation '{}'".format(device, key, validation)) + # Warn once per register, this runs on every settings refresh + if (device, key) not in self.validation_parse_warned: + self.log("GECloud: Warn: Unable to parse options for setting {} {} from validation '{}'".format(device, key, validation)) + self.validation_parse_warned.add((device, key)) else: options_text = validation_options diff --git a/apps/predbat/tests/test_ge_cloud.py b/apps/predbat/tests/test_ge_cloud.py index 7a257e8c9..28266ae6f 100644 --- a/apps/predbat/tests/test_ge_cloud.py +++ b/apps/predbat/tests/test_ge_cloud.py @@ -73,6 +73,7 @@ def __init__(self): self.ems_device = None self.gateway_device = None self.ems_slot_warned = set() + self.validation_parse_warned = set() self._discovery_report = None self._now_utc_exact = datetime.now(timezone.utc) self.settings_from_cache = False @@ -4064,6 +4065,28 @@ async def mock_publish(*args, **kwargs): print("ERROR: Expected a warning for unparseable validation, got {}".format(ge_cloud.log_messages)) return 1 + # An empty option list must not parse as [""] and reject every value; it falls back to 'in:' too + ge_cloud.settings["test123"][90] = {"value": "1", "validation": "Value must be one of: ()", "validation_rules": ["in:1,2"]} + await ge_cloud.select_event(entity_id, "1") + if write_calls != ["Export (AC3)", "2", "1"]: + print("ERROR: Expected empty option list to fall back to 'in:' and accept '1', got {}".format(write_calls)) + return 1 + + # A non-string validation value is ignored rather than raising AttributeError + ge_cloud.settings["test123"][90] = {"value": "1", "validation": {"unexpected": True}, "validation_rules": ["in:1,2"]} + await ge_cloud.select_event(entity_id, "2") + if write_calls != ["Export (AC3)", "2", "1", "2"]: + print("ERROR: Expected non-string validation to fall back to 'in:' and accept '2', got {}".format(write_calls)) + return 1 + + # Unparseable validation with no 'in:' rule has nothing to check against, so the write goes through + # and the GE Cloud API validates it server-side (the same as any select without option validation text) + ge_cloud.settings["test123"][90] = {"value": "1", "validation": "Value must be one of:", "validation_rules": []} + await ge_cloud.select_event(entity_id, "5") + if write_calls != ["Export (AC3)", "2", "1", "2", "5"]: + print("ERROR: Expected write without local validation when no options are known, got {}".format(write_calls)) + return 1 + return 0 return run_async(test()) @@ -4076,6 +4099,13 @@ def _test_parse_validation_options(my_predbat): cases = [ ("Value must be one of: (00:00, 00:30, 01:00)", ["00:00", "00:30", "01:00"]), ("Value must be one of: (Eco (Paused), Timed Demand, Export (AC3))", ["Eco (Paused)", "Timed Demand", "Export (AC3)"]), + # Commas inside a label's own brackets do not split it + ("Value must be one of: (Eco, Pause (Battery, Grid), Export)", ["Eco", "Pause (Battery, Grid)", "Export"]), + ("Value must be one of: (0,1,2)", ["0", "1", "2"]), + # An empty option list is unparseable, not a single empty-string option + ("Value must be one of: ()", None), + ("Value must be one of: (", None), + ("Value must be one of: ( )", None), ("Value must be one of:", None), ("Value must be between 0 and 100", None), ("", None), @@ -4334,6 +4364,14 @@ def _test_publish_registers_bracketed_options(my_predbat): print("ERROR: Expected later registers to still be published") return 1 + # A permanently unparseable register warns once, not on every settings refresh + run_async(ge_cloud.publish_registers("test123", registers)) + run_async(ge_cloud.publish_registers("test123", registers)) + warnings = [message for message in ge_cloud.log_messages if "Unable to parse options" in message] + if len(warnings) != 1: + print("ERROR: Expected exactly one unparseable-validation warning across three refreshes, got {}".format(warnings)) + return 1 + return 0