feat: prefer unified warning about equipment with repeated names, with context about source of equipment

pyomnilogic_local/collections.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import logging
-from collections import Counter
 from enum import Enum
 from typing import TYPE_CHECKING, Any, overload
 
@@ -15,9 +14,6 @@
 
 _LOGGER = logging.getLogger(__name__)
 
-# Track which duplicate names we've already warned about to avoid log spam
-_WARNED_DUPLICATE_NAMES: set[str] = set()
-
 
 class EquipmentDict[OE: OmniEquipment[Any, Any]]:
     """A dictionary-like collection that supports lookup by both name and system_id.
@@ -72,10 +68,6 @@ def __init__(self, items: list[OE] | None = None) -> None:
     def _validate(self) -> None:
         """Validate the equipment collection.
 
-        Checks for:
-        1. Items without both system_id and name (raises ValueError)
-        2. Duplicate names (logs warning once per unique duplicate)
-
         Raises:
             ValueError: If any item has neither a system_id nor a name.
         """
@@ -88,23 +80,6 @@ def _validate(self) -> None:
             )
             raise ValueError(msg)
 
-        # Find duplicate names that we haven't warned about yet
-        name_counts = Counter(item.name for item in self._items if item.name is not None)
-        duplicate_names = {name for name, count in name_counts.items() if count > 1}
-        unwarned_duplicates = duplicate_names.difference(_WARNED_DUPLICATE_NAMES)
-
-        # Log warnings for new duplicates
-        for name in unwarned_duplicates:
-            _LOGGER.warning(
-                "Equipment collection contains %d items with the same name '%s'. "
-                "Name-based lookups will return the first match. "
-                "Consider using system_id-based lookups for reliability "
-                "or renaming equipment to avoid duplicates.",
-                name_counts[name],
-                name,
-            )
-            _WARNED_DUPLICATE_NAMES.add(name)
-
     @property
     def _by_name(self) -> dict[str, OE]:
         """Dynamically build name-to-equipment mapping."""

pyomnilogic_local/omnilogic.py

@@ -16,6 +16,8 @@
 from pyomnilogic_local.system import System
 
 if TYPE_CHECKING:
+    from collections.abc import Sequence
+
     from pyomnilogic_local._base import OmniEquipment
     from pyomnilogic_local.bow import Bow
     from pyomnilogic_local.chlorinator import Chlorinator
@@ -35,6 +37,29 @@
 _LOGGER = logging.getLogger(__name__)
 
 
+def _check_duplicate_item_names(items: Sequence[OmniEquipment[Any, Any]], source_id: str) -> str | None:
+    """Returns a warning message if there are items with a duplicate name."""
+    items_by_name: dict[str, list[OmniEquipment[Any, Any]]] = {}
+    for item in items:
+        if item.name is not None:
+            items_by_name.setdefault(item.name, []).append(item)
+
+    duplicate_names = {name for name, name_items in items_by_name.items() if len(name_items) > 1}
+    if not duplicate_names:
+        return None
+
+    duplicates = []
+    for name in sorted(duplicate_names):
+        ids = [i.system_id for i in items_by_name[name] if i.system_id is not None]
+        duplicates.append(f"'{name}' {ids}")
+    return (
+        f"OmniLogic {source_id} provided equipment with duplicate names: {', '.join(duplicates)}. "
+        "Name-based lookups will return the first match. "
+        "Consider looking up by system_id (shown in parentheses) for reliability "
+        "or renaming equipment on the OmniLogic controller to avoid duplicates."
+    )
+
+
 class OmniLogic:
     mspconfig: MSPConfig
     telemetry: Telemetry
@@ -65,6 +90,7 @@ def __init__(self, host: str, port: int = 10444, timeout: float = DEFAULT_RESPON
         else:
             self._api = OmniLogicAPI(host, port, timeout)
         self._refresh_lock = asyncio.Lock()
+        self._seen_item_names: set[str] = set()
 
     def __repr__(self) -> str:
         """Return a string representation of the OmniLogic instance for debugging.
@@ -182,6 +208,12 @@ def _update_equipment(self) -> None:
         else:
             self.schedules = EquipmentDict([Schedule(self, schedule_, self.telemetry) for schedule_ in self.mspconfig.schedules])
 
+        current_names = {item.name for item in self.all_equipment if item.name is not None}
+        if not current_names.issubset(self._seen_item_names):
+            if warning := _check_duplicate_item_names(self.all_equipment, f"{self.host}:{self.port}"):
+                _LOGGER.warning(warning)
+            self._seen_item_names.update(current_names)
+
     # Equipment discovery properties
     @property
     def all_lights(self) -> EquipmentDict[ColorLogicLight]:
@@ -276,6 +308,26 @@ def all_csads(self) -> EquipmentDict[CSAD]:
             csads.extend(bow.csads.values())
         return EquipmentDict(csads)
 
+    @property
+    def all_equipment(self) -> list[OmniEquipment[Any, Any]]:
+        """Returns a flat list of all equipment instances in the system."""
+        equipment: list[OmniEquipment[Any, Any]] = [self.backyard]
+        equipment.extend(self.all_lights.values())
+        equipment.extend(self.all_relays.values())
+        equipment.extend(self.all_pumps.values())
+        equipment.extend(self.all_filters.values())
+        equipment.extend(self.all_sensors.values())
+        equipment.extend(self.all_heaters.values())
+        equipment.extend(self.all_heater_equipment.values())
+        equipment.extend(self.all_chlorinators.values())
+        equipment.extend(self.all_chlorinator_equipment.values())
+        equipment.extend(self.all_csads.values())
+        equipment.extend(self.all_csad_equipment.values())
+        equipment.extend(self.all_bows.values())
+        equipment.extend(self.groups.values())
+        equipment.extend(self.schedules.values())
+        return equipment
+
     @property
     def all_bows(self) -> EquipmentDict[Bow]:
         """Returns all Bow instances across all bows in the backyard."""
@@ -292,24 +344,7 @@ def get_equipment_by_name(self, name: str) -> OmniEquipment[Any, Any] | None:
         Returns:
             The first equipment with matching name, or None if not found
         """
-        # Search all equipment types
-        all_equipment: list[OmniEquipment[Any, Any]] = []
-        all_equipment.extend([self.backyard])
-        all_equipment.extend(self.all_lights.values())
-        all_equipment.extend(self.all_relays.values())
-        all_equipment.extend(self.all_pumps.values())
-        all_equipment.extend(self.all_filters.values())
-        all_equipment.extend(self.all_sensors.values())
-        all_equipment.extend(self.all_heaters.values())
-        all_equipment.extend(self.all_heater_equipment.values())
-        all_equipment.extend(self.all_chlorinators.values())
-        all_equipment.extend(self.all_chlorinator_equipment.values())
-        all_equipment.extend(self.all_csads.values())
-        all_equipment.extend(self.all_csad_equipment.values())
-        all_equipment.extend(self.all_bows.values())
-        all_equipment.extend(self.groups.values())
-
-        for equipment in all_equipment:
+        for equipment in self.all_equipment:
             if equipment.name == name:
                 return equipment
 
@@ -324,25 +359,7 @@ def get_equipment_by_id(self, system_id: int) -> OmniEquipment[Any, Any] | None:
         Returns:
             The first equipment with matching system_id, or None if not found
         """
-        # Search all equipment types
-        all_equipment: list[OmniEquipment[Any, Any]] = []
-        all_equipment.extend([self.backyard])
-        all_equipment.extend(self.all_lights.values())
-        all_equipment.extend(self.all_relays.values())
-        all_equipment.extend(self.all_pumps.values())
-        all_equipment.extend(self.all_filters.values())
-        all_equipment.extend(self.all_sensors.values())
-        all_equipment.extend(self.all_heaters.values())
-        all_equipment.extend(self.all_heater_equipment.values())
-        all_equipment.extend(self.all_chlorinators.values())
-        all_equipment.extend(self.all_chlorinator_equipment.values())
-        all_equipment.extend(self.all_csads.values())
-        all_equipment.extend(self.all_csad_equipment.values())
-        all_equipment.extend(self.all_bows.values())
-        all_equipment.extend(self.groups.values())
-        all_equipment.extend(self.schedules.values())
-
-        for equipment in all_equipment:
+        for equipment in self.all_equipment:
             if equipment.system_id == system_id:
                 return equipment
 

tests/test_omnilogic.py

@@ -0,0 +1,60 @@
+from typing import Any
+
+from pyomnilogic_local.omnilogic import _check_duplicate_item_names
+
+
+class FakeEquipment:
+    """A fake equipment class for testing duplicate detection."""
+
+    def __init__(self, system_id: int, name: str) -> None:
+        self.system_id = system_id
+        self.name = name
+
+
+def test__check_duplicate_item_names() -> None:
+    """Test the duplicate name warning helper directly."""
+    items: Any = [
+        FakeEquipment(1, "Solar"),
+        FakeEquipment(2, "Solar"),
+        FakeEquipment(3, "Gas"),
+        FakeEquipment(4, "Gas"),
+    ]
+    warning = _check_duplicate_item_names(
+        items,
+        source_id="127.0.0.1:10444",
+    )
+
+    assert warning == (
+        "OmniLogic 127.0.0.1:10444 provided equipment with duplicate names: 'Gas' [3, 4], 'Solar' [1, 2]. "
+        "Name-based lookups will return the first match. "
+        "Consider looking up by system_id (shown in parentheses) for reliability "
+        "or renaming equipment on the OmniLogic controller to avoid duplicates."
+    )
+
+
+def test__check_duplicate_item_names_different_host() -> None:
+    """Test the duplicate name warning helper with a different host."""
+    items: Any = [
+        FakeEquipment(1, "Solar"),
+        FakeEquipment(2, "Solar"),
+    ]
+    warning = _check_duplicate_item_names(
+        items,
+        source_id="127.0.0.2:3000",
+    )
+
+    assert warning is not None
+    assert "OmniLogic 127.0.0.2:3000 provided equipment with duplicate names:" in warning
+
+
+def test__check_duplicate_item_names_no_duplicates() -> None:
+    """Test that the helper returns None when there are no duplicates."""
+    items: Any = [
+        FakeEquipment(1, "Solar"),
+        FakeEquipment(2, "Gas"),
+    ]
+    warning = _check_duplicate_item_names(
+        items,
+        source_id="127.0.0.1:10444",
+    )
+    assert warning is None