among matched inverters, favour the least permissive schema

solax/discovery.py

@@ -3,7 +3,7 @@
 import sys
 from asyncio import Future, Task
 from collections import defaultdict
-from typing import Dict, Literal, Sequence, Set, Type, TypedDict, Union, cast
+from typing import Dict, Literal, Sequence, Set, Tuple, Type, TypedDict, Union, cast
 
 from solax.inverter import Inverter
 from solax.inverter_http_client import InverterHttpClient
@@ -13,19 +13,69 @@
 if sys.version_info >= (3, 10):
     from importlib.metadata import entry_points
 else:
-    from importlib_metadata import entry_points
+    from importlib_metadata import entry_points  # pragma: no cover
 
 if sys.version_info >= (3, 11):
     from typing import Unpack
 else:
-    from typing_extensions import Unpack
+    from typing_extensions import Unpack  # pragma: no cover
+
+# Ranks inverter schemas from most-specific (least permissive) to
+# least-specific (most permissive). REGISTRY is sorted by this table so
+# that discover()'s tie-break (lowest rank wins) prefers the more
+# tightly matching inverter when several schemas validate the same
+# response.
+#
+# This is NOT derived from entry_points.txt order: setuptools
+# alphabetizes entry points on install , so that ordering can't be
+# relied on. This table is maintained by hand; drift is caught by
+# tests/test_schema_ambiguity.py::test_registry_order_matches_specificity,
+# which recomputes the expected order from tests/fixtures.py.
+#
+# Classes not listed here (e.g. a newly added inverter) rank last -- a
+# conservative default that never lets an unranked schema outrank a
+# known-specific one. The guardrail test forces new inverters to be
+# ranked explicitly rather than silently relying on this default.
+_SCHEMA_SPECIFICITY_ORDER: Tuple[str, ...] = (
+    "X3Ultra",
+    "X3MicProG2",
+    "X3EVC",
+    "X1Mini",
+    "QVOLTHYBG33P",
+    "XHybrid",
+    "X3HybridG4",
+    "X1Smart",
+    "X1LiteLV",
+    "X1HybridGen4",
+    "X1G4Series",
+    "X1",
+    "X3V34",
+    "X3",
+    "X1MiniV34",
+    "X1Boost",
+)
+_SCHEMA_SPECIFICITY_RANK: Dict[str, int] = {
+    name: rank for rank, name in enumerate(_SCHEMA_SPECIFICITY_ORDER)
+}
+
+
+def _specificity_rank(inverter_class: Type[Inverter]) -> int:
+    return _SCHEMA_SPECIFICITY_RANK.get(
+        inverter_class.__name__, len(_SCHEMA_SPECIFICITY_ORDER)
+    )
+
 
 # registry of inverters
-REGISTRY: Set[Type[Inverter]] = {
-    ep.load()
-    for ep in entry_points(group="solax.inverter")
-    if issubclass(ep.load(), Inverter)
-}
+REGISTRY: Tuple[Type[Inverter], ...] = tuple(
+    sorted(
+        dict.fromkeys(
+            loaded
+            for ep in entry_points(group="solax.inverter")
+            if issubclass(loaded := ep.load(), Inverter)
+        ),
+        key=_specificity_rank,
+    )
+)
 
 logging.basicConfig(level=logging.INFO)
 
@@ -38,7 +88,7 @@ class DiscoveryKeywords(TypedDict, total=False):
 if sys.version_info >= (3, 9):
     _InverterTask = Task[Inverter]
 else:
-    _InverterTask = Task
+    _InverterTask = Task  # pragma: no cover
 
 
 class _DiscoveryHttpClient:
@@ -73,15 +123,21 @@ async def _discovery_task(i) -> Inverter:
 async def discover(
     host, port, pwd="", **kwargs: Unpack[DiscoveryKeywords]
 ) -> Union[Inverter, Set[Inverter]]:
+    # pylint: disable=too-many-locals
     done: Set[_InverterTask] = set()
     pending: Set[_InverterTask] = set()
     failures = set()
     requests: Dict[InverterHttpClient, Future] = defaultdict(
         asyncio.get_running_loop().create_future
     )
 
+    # rank of the class each task was built from, in the order it was
+    # offered to discover(); used to break ties deterministically when
+    # several inverters' schemas match the same response
+    priority: Dict[_InverterTask, int] = {}
+
     return_when = kwargs.get("return_when", asyncio.FIRST_COMPLETED)
-    for cls in kwargs.get("inverters", REGISTRY):
+    for rank, cls in enumerate(kwargs.get("inverters", REGISTRY)):
         for inverter in cls.build_all_variants(host, port, pwd):
             inverter.http_client = cast(
                 InverterHttpClient,
@@ -90,9 +146,9 @@ async def discover(
                 ),
             )
 
-            pending.add(
-                asyncio.create_task(_discovery_task(inverter), name=f"{inverter}")
-            )
+            task = asyncio.create_task(_discovery_task(inverter), name=f"{inverter}")
+            priority[task] = rank
+            pending.add(task)
 
     if not pending:
         raise DiscoveryError("No inverters to try to discover")
@@ -143,7 +199,8 @@ async def stagger() -> None:
     if done:
         logging.info("Discovered inverters: %s", {task.result() for task in done})
         if return_when == asyncio.FIRST_COMPLETED:
-            return await next(iter(done))
+            winner = min(done, key=priority.__getitem__)
+            return await winner
 
         return {task.result() for task in done}
 

solax/inverter.py

@@ -91,7 +91,7 @@ def sensor_map(cls) -> Dict[str, Tuple[int, Measurement]]:
         for name, mapping in cls.response_decoder().items():
             unit = Measurement(Units.NONE)
 
-            (idx, unit_or_measurement, *_) = mapping
+            idx, unit_or_measurement, *_ = mapping
 
             if isinstance(unit_or_measurement, Units):
                 unit = Measurement(unit_or_measurement)

solax/inverter_http_client.py

@@ -11,7 +11,7 @@
 
 __all__ = ("InverterHttpClient", "Method")
 
-if sys.version_info >= (3, 10):
+if sys.version_info >= (3, 10):  # pragma: no branch
     from dataclasses import KW_ONLY
 
 
@@ -26,7 +26,7 @@ class Method(Enum):
 
 _kwargs: Dict[str, bool] = {}
 
-if sys.version_info >= (3, 11):
+if sys.version_info >= (3, 11):  # pragma: no branch
     _kwargs["slots"] = True
     _kwargs["weakref_slot"] = True
 
@@ -35,7 +35,7 @@ class Method(Enum):
 class InverterHttpClient:
     """Initialize the Http client."""
 
-    if sys.version_info >= (3, 10):
+    if sys.version_info >= (3, 10):  # pragma: no branch
         _: KW_ONLY
 
     url: str

solax/inverters/x1_lite_lv.py

@@ -11,7 +11,7 @@ class X1LiteLV(Inverter):
     # pylint: disable=duplicate-code
     _schema = vol.Schema(
         {
-            vol.Required("type"): int,
+            vol.Required("type"): vol.All(int, 103),
             vol.Required("sn"): str,
             vol.Required("ver"): str,
             vol.Required("data"): vol.Schema(

solax/inverters/x_hybrid.py

@@ -16,7 +16,7 @@ class XHybrid(Inverter):
         {
             vol.Required("method"): str,
             vol.Required("version"): str,
-            vol.Required("type"): str,
+            vol.Required("type"): vol.All(str, "AL_SE"),
             vol.Required("sn"): str,
             vol.Required("data"): vol.Schema(
                 vol.All(

solax/response_parser.py

@@ -16,7 +16,7 @@
 if sys.version_info >= (3, 11):
     from typing import Unpack
 else:
-    from typing_extensions import Unpack
+    from typing_extensions import Unpack  # pragma: no cover
 
 _LOGGER = logging.getLogger(__name__)
 _LOGGER.setLevel(logging.INFO)
@@ -46,7 +46,7 @@ def serial_number(self):
 _KEY_TYPE = "type"
 
 
-GenericResponseSchema = vol.All(
+GENERIC_RESPONSE_SCHEMA = vol.All(
     vol.Schema({vol.Required(_KEY_SERIAL): str}, extra=vol.ALLOW_EXTRA),
     vol.Any(
         vol.Schema({vol.Required(_KEY_VERSION): str}, extra=vol.ALLOW_EXTRA),
@@ -77,7 +77,7 @@ def __init__(
         dongle_serial_number_getter: Callable[[Dict[str, Any]], Optional[str]],
         inverter_serial_number_getter: Callable[[Dict[str, Any]], Optional[str]],
     ) -> None:
-        self.schema = vol.And(GenericResponseSchema, schema)
+        self.schema = vol.And(GENERIC_RESPONSE_SCHEMA, schema)
         self.response_decoder = decoder
         self.dongle_serial_number_getter = dongle_serial_number_getter
         self.inverter_serial_number_getter = inverter_serial_number_getter
@@ -95,7 +95,7 @@ def _postprocess_gen(
         Return map of functions to be applied to each sensor value
         """
         for name, mapping in self.response_decoder.items():
-            (_, _, *processors) = mapping
+            _, _, *processors = mapping
             for processor in processors:
                 yield name, processor
 

tests/test_schema_ambiguity.py

@@ -0,0 +1,111 @@
+"""
+Checks that no two registered inverters accept the same response.
+
+Also runnable standalone to print inverter schemas ranked from most to
+least permissive:
+
+    python -m tests.test_schema_ambiguity
+"""
+
+from collections import defaultdict
+
+import pytest
+import voluptuous as vol
+from voluptuous import Invalid, MultipleInvalid
+
+from solax.discovery import REGISTRY
+from solax.response_parser import GENERIC_RESPONSE_SCHEMA
+from tests import fixtures
+
+
+def _matching_inverters(response):
+    normalized = {key.lower(): value for key, value in response.items()}
+    matches = set()
+    for inverter_class in REGISTRY:
+        combined_schema = vol.And(GENERIC_RESPONSE_SCHEMA, inverter_class.schema())
+        try:
+            combined_schema(dict(normalized))
+        except (Invalid, MultipleInvalid):
+            continue
+        matches.add(inverter_class)
+    return matches
+
+
+def _permissiveness_scores():
+    """
+    Count, for each inverter class, how many fixture responses (belonging
+    to any inverter) its schema accepts. A schema that accepts more
+    unrelated responses is less specific (more permissive) than one that
+    accepts fewer.
+    """
+    scores = defaultdict(int)
+    for case in fixtures.INVERTERS_UNDER_TEST:
+        for inverter_class in _matching_inverters(case.response):
+            scores[inverter_class] += 1
+    return scores
+
+
+_PERMISSIVENESS_SCORES = _permissiveness_scores()
+
+
+def _most_permissive_first(inverter_classes):
+    return sorted(
+        inverter_classes,
+        key=lambda c: (-_PERMISSIVENESS_SCORES[c], c.__name__),
+    )
+
+
+@pytest.mark.xfail(
+    reason="some inverter schemas are still too permissive; tightening them "
+    "is tracked separately and shouldn't block CI in the meantime",
+    strict=False,
+)
+@pytest.mark.parametrize(
+    "case",
+    fixtures.INVERTERS_UNDER_TEST,
+    ids=[
+        f"{i}-{case.inverter.__name__}"
+        for i, case in enumerate(fixtures.INVERTERS_UNDER_TEST)
+    ],
+)
+def test_response_matches_exactly_one_inverter_schema(case):
+    matches = _matching_inverters(case.response)
+    extra = matches - {case.inverter}
+    assert matches == {case.inverter}, (
+        f"expected only {case.inverter.__name__} to accept this response, "
+        "but it also validated against (most to least permissive): "
+        f"{[c.__name__ for c in _most_permissive_first(extra)]}"
+    )
+
+
+def test_registry_order_matches_specificity():
+    """
+    REGISTRY must be ordered least-to-most permissive so discover()'s
+    tie-break prefers the more specific/correct inverter. This is a
+    drift guard: if fixtures or schemas change such that the
+    fixture-computed ranking no longer matches REGISTRY's actual order,
+    update _SCHEMA_SPECIFICITY_ORDER in solax/discovery.py to match
+    (rerun `python -m tests.test_schema_ambiguity` and reverse it, or
+    use _most_permissive_first(REGISTRY) directly).
+    """
+    expected = list(reversed(_most_permissive_first(REGISTRY)))
+    expected_literal = ",\n".join(f'    "{c.__name__}"' for c in expected)
+    assert REGISTRY == tuple(expected), (
+        "solax.discovery._SCHEMA_SPECIFICITY_ORDER has drifted from the "
+        "fixture-computed permissiveness ranking; update it to:\n"
+        f"{expected_literal}"
+    )
+
+
+def _print_permissiveness_ranking():
+    ambiguous = [c for c in REGISTRY if _PERMISSIVENESS_SCORES[c] > 1]
+    ordered = _most_permissive_first(ambiguous)
+    width = max(len(c.__name__) for c in ordered)
+    for rank, inverter_class in enumerate(ordered, start=1):
+        name = inverter_class.__name__
+        matches = _PERMISSIVENESS_SCORES[inverter_class]
+        print(f"{rank:>2}. {name:<{width}}  matches={matches}")
+
+
+if __name__ == "__main__":
+    _print_permissiveness_ranking()