From a714424e5953fd55b2205a936e71564c38dee12b Mon Sep 17 00:00:00 2001 From: Matt Brown Date: Sat, 25 Apr 2026 00:07:25 -0700 Subject: [PATCH 1/3] 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. ``` --- pyomnilogic_local/collections.py | 25 --------- pyomnilogic_local/omnilogic.py | 94 +++++++++++++++++++------------- tests/test_omnilogic.py | 62 +++++++++++++++++++++ 3 files changed, 119 insertions(+), 62 deletions(-) create mode 100644 tests/test_omnilogic.py diff --git a/pyomnilogic_local/collections.py b/pyomnilogic_local/collections.py index be1f39b..62f3986 100644 --- a/pyomnilogic_local/collections.py +++ b/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.""" diff --git a/pyomnilogic_local/omnilogic.py b/pyomnilogic_local/omnilogic.py index 13911fd..5be1e54 100644 --- a/pyomnilogic_local/omnilogic.py +++ b/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 diff --git a/tests/test_omnilogic.py b/tests/test_omnilogic.py new file mode 100644 index 0000000..99f60c0 --- /dev/null +++ b/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 From 2c972a91173209cbf080ea4b8e39df6989b7eca4 Mon Sep 17 00:00:00 2001 From: Matt Brown Date: Sat, 25 Apr 2026 15:53:45 -0700 Subject: [PATCH 2/3] fix: update source_id in duplicate item name test to private address, to address SonarQube feedback, and remove stale line --- tests/test_omnilogic.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_omnilogic.py b/tests/test_omnilogic.py index 99f60c0..5ac4e42 100644 --- a/tests/test_omnilogic.py +++ b/tests/test_omnilogic.py @@ -42,11 +42,11 @@ def test__check_duplicate_item_names_different_host() -> None: ] warning = _check_duplicate_item_names( items, - source_id="10.0.0.1:3000", + source_id="127.0.0.2:3000", ) assert warning is not None - assert "OmniLogic 10.0.0.1:3000 contains multiple items with the same name:" in warning + assert "OmniLogic 127.0.0.2:3000 contains multiple items with the same name:" in warning def test__check_duplicate_item_names_no_duplicates() -> None: From a00b4b028ef6c5ba1f9f62da2c4d0c8d5a391cf2 Mon Sep 17 00:00:00 2001 From: Matt Brown Date: Tue, 28 Apr 2026 00:46:59 -0700 Subject: [PATCH 3/3] refactor: condense duplicate item name warning message into a single, more compact line --- pyomnilogic_local/omnilogic.py | 15 ++++++--------- tests/test_omnilogic.py | 8 +++----- 2 files changed, 9 insertions(+), 14 deletions(-) diff --git a/pyomnilogic_local/omnilogic.py b/pyomnilogic_local/omnilogic.py index 5be1e54..ccb3ea3 100644 --- a/pyomnilogic_local/omnilogic.py +++ b/pyomnilogic_local/omnilogic.py @@ -48,19 +48,16 @@ def _check_duplicate_item_names(items: Sequence[OmniEquipment[Any, Any]], source 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:"] + duplicates = [] 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( + 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 using system_id-based lookups for reliability " + "Consider looking up by system_id (shown in parentheses) for reliability " "or renaming equipment on the OmniLogic controller to avoid duplicates." ) - return "\n".join(msg_lines) class OmniLogic: diff --git a/tests/test_omnilogic.py b/tests/test_omnilogic.py index 5ac4e42..f56778e 100644 --- a/tests/test_omnilogic.py +++ b/tests/test_omnilogic.py @@ -25,11 +25,9 @@ def test__check_duplicate_item_names() -> None: ) 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" + "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 using system_id-based lookups for reliability " + "Consider looking up by system_id (shown in parentheses) for reliability " "or renaming equipment on the OmniLogic controller to avoid duplicates." ) @@ -46,7 +44,7 @@ def test__check_duplicate_item_names_different_host() -> None: ) assert warning is not None - assert "OmniLogic 127.0.0.2:3000 contains multiple items with the same name:" in warning + assert "OmniLogic 127.0.0.2:3000 provided equipment with duplicate names:" in warning def test__check_duplicate_item_names_no_duplicates() -> None: