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

Check for duplicate names in the main refresh loop. Describe that they
are due to output from the OmniLogic unit itself. This will help clarify
the source in downstream usage, like in Home Assistant (HA).

I personally saw messages in HA like the following and thought I had
incorrectly set up the HA OmniLogic integration:

```
2026-04-24 20:56:23.095 WARNING (MainThread) [pyomnilogic_local.collections] Equipment collection contains 4 items with the same name 'SolarSensor'. Name-based lookups will return the first match. Consider using system_id-based lookups for reliability or renaming equipment to avoid duplicates.
2026-04-24 20:56:24.636 WARNING (MainThread) [pyomnilogic_local.collections] Equipment collection contains 2 items with the same name 'Gas'. Name-based lookups will return the first match. Consider using system_id-based lookups for reliability or renaming equipment to avoid duplicates.
2026-04-24 20:56:24.637 WARNING (MainThread) [pyomnilogic_local.collections] Equipment collection contains 2 items with the same name 'Solar'. Name-based lookups will return the first match. Consider using system_id-based lookups for reliability or renaming equipment to avoid duplicates.
2026-04-24 20:56:24.638 WARNING (MainThread) [pyomnilogic_local.collections] Equipment collection contains 2 items with the same name 'Filter Pump'. Name-based lookups will return the first match. Consider using system_id-based lookups for reliability or renaming equipment to avoid duplicates.
2026-04-24 20:56:24.639 WARNING (MainThread) [pyomnilogic_local.collections] Equipment collection contains 2 items with the same name 'Chlorinator'. Name-based lookups will return the first match. Consider using system_id-based lookups for reliability or renaming equipment to avoid duplicates.
2026-04-24 20:56:24.641 WARNING (MainThread) [pyomnilogic_local.collections] Equipment collection contains 2 items with the same name 'WaterSensor'. Name-based lookups will return the first match. Consider using system_id-based lookups for reliability or renaming equipment to avoid duplicates.
2026-04-24 20:56:24.641 WARNING (MainThread) [pyomnilogic_local.collections] Equipment collection contains 2 items with the same name 'FlowSensor'. Name-based lookups will return the first match. Consider using system_id-based lookups for reliability or renaming equipment to avoid duplicates.
```

Author: m-d-brown

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,32 @@
 _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
+
+    item = items[0]  # Guaranteed to exist if we have duplicates
+    msg_lines = [f"OmniLogic {source_id} contains multiple items with the same name:"]
+    for name in sorted(duplicate_names):
+        name_items = items_by_name[name]
+        ids = [str(i.system_id) for i in name_items if i.system_id is not None]
+        msg_lines.append(f"  - {name}: IDs {', '.join(ids)}")
+
+    msg_lines.append(
+        "Name-based lookups will return the first match. "
+        "Consider using system_id-based lookups for reliability "
+        "or renaming equipment on the OmniLogic controller to avoid duplicates."
+    )
+    return "\n".join(msg_lines)
+
+
 class OmniLogic:
     mspconfig: MSPConfig
     telemetry: Telemetry
@@ -65,6 +93,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 +211,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 +311,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 +347,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 +362,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,62 @@
+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 contains multiple items with the same name:\n"
+        "  - Gas: IDs 3, 4\n"
+        "  - Solar: IDs 1, 2\n"
+        "Name-based lookups will return the first match. "
+        "Consider using system_id-based lookups 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="10.0.0.1:3000",
+    )
+
+    assert warning is not None
+    assert "OmniLogic 10.0.0.1:3000 contains multiple items with the same name:" 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