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..ccb3ea3 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,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 diff --git a/tests/test_omnilogic.py b/tests/test_omnilogic.py new file mode 100644 index 0000000..f56778e --- /dev/null +++ b/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