Merge pull request #20 from dala318/tests

Add some basic but relevant tests

custom_components/ev_load_balancing/chargers/virtual.py

@@ -0,0 +1,144 @@
+"""Handling Virtual Charger."""
+
+import logging
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.event import async_track_state_change_event
+from homeassistant.helpers.template import device_entities
+
+from ..const import Phases
+from ..helpers.entity_value import get_sensor_entity_attribute_value
+from . import Charger, ChargerPhase, ChargingState
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class ChargerPhaseVirtual(ChargerPhase):
+    """A data class for a charger phase."""
+
+    def __init__(self, hass: HomeAssistant, entity_id: str, attribute: str) -> None:
+        """Initialize object."""
+        self._hass = hass
+        self._entity = entity_id
+        self._attribute = attribute
+        self._value = None
+
+    def set_expected(self, value: float) -> None:
+        """Fake update mathod."""
+        self._value = value
+
+    def update(self) -> None:
+        """Do nothing, but exists to conform to standard."""
+
+    def current_limit(self) -> float:
+        """Get set current limit on phase."""
+        return self._value
+
+
+class ChargerVirtual(Charger):
+    """Slimmelezer mains extractor."""
+
+    _state_change_listeners = []
+
+    def __init__(
+        self, hass: HomeAssistant, update_callback, device_id: str, ttl: int
+    ) -> None:
+        """Initilalize Slimmelezer extractor."""
+        super().__init__(hass, update_callback)
+        self._id = device_id
+        self._ttl = ttl
+
+        # entities = device_entities(hass, device_id)
+
+        # self._ent_status = [e for e in entities if e.endswith("_status")][0]
+        # self._state_change_listeners.append(
+        #     async_track_state_change_event(
+        #         self._hass,
+        #         [self._ent_status],
+        #         self._async_input_changed,
+        #     )
+        # )
+
+        # self._ent_circuit_limit = [
+        #     e for e in entities if e.endswith("_dynamic_circuit_limit")
+        # ][0]
+        # self._state_change_listeners.append(
+        #     async_track_state_change_event(
+        #         self._hass,
+        #         [self._ent_circuit_limit],
+        #         self._async_input_changed,
+        #     )
+        # )
+
+        # self._phase1 = ChargerPhaseVirtual(
+        #     self._hass, self._ent_circuit_limit, "state_dynamicCircuitCurrentP1"
+        # )
+        # self._phase2 = ChargerPhaseVirtual(
+        #     self._hass, self._ent_circuit_limit, "state_dynamicCircuitCurrentP2"
+        # )
+        # self._phase3 = ChargerPhaseVirtual(
+        #     self._hass, self._ent_circuit_limit, "state_dynamicCircuitCurrentP3"
+        # )
+
+    def set_expected(self, current_limits, set_limits) -> None:
+        """Set expected values."""
+
+    async def async_set_limits(
+        self, phase1: float, phase2: float, phase3: float
+    ) -> bool:
+        """Set charger limits."""
+        _LOGGER.debug(
+            "Setting limits: phase 1 %f, phase 2 %f, phase 3 %f", phase1, phase2, phase3
+        )
+        domain = "easee"
+        service = "set_circuit_dynamic_limit"
+        service_data = {
+            "device_id": self._id,
+            "current_p1": phase1,
+            "current_p2": phase2,
+            "current_p3": phase3,
+            "time_to_live": self._ttl,
+        }
+        # await self._hass.services.async_call(domain, service, service_data)
+
+        return True
+
+    def update(self) -> None:
+        """Update measuremetns."""
+        self._phase1.update()
+        self._phase2.update()
+        self._phase3.update()
+
+    def cleanup(self):
+        """Cleanup by removing event listeners."""
+        # for listner in self._state_change_listeners:
+        #     listner()
+
+    @property
+    def charging_state(self) -> ChargingState:
+        """Return if charging state."""
+        if self._hass.states.get(self._ent_status).state in ["charging"]:
+            return ChargingState.CHARGING
+        if self._hass.states.get(self._ent_status).state in ["awaiting_start"]:
+            return ChargingState.PENDING
+        return ChargingState.OFF
+
+    def get_phase(self, phase: Phases) -> ChargerPhase:
+        """Return phase X data."""
+        if phase == Phases.PHASE1:
+            return self._phase1
+        if phase == Phases.PHASE2:
+            return self._phase2
+        if phase == Phases.PHASE3:
+            return self._phase3
+        return None
+
+    def get_rated_limit(self) -> int:
+        """Return overall limit per phase on charger circuit."""
+        limit = get_sensor_entity_attribute_value(
+            self._hass, _LOGGER, self._ent_circuit_limit, "circuit_ratedCurrent"
+        )
+        if limit is not None:
+            limit = int(limit)
+            _LOGGER.debug("Returning rated limit %d for charger circuit", limit)
+        return limit

custom_components/ev_load_balancing/config_flow.py

@@ -54,6 +54,19 @@
     NAME_TEMPLATE: MainsTemplate,
 }
 
+USER_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_NAME): str,
+        vol.Required(CONF_MAINS_TYPE): vol.In(
+            _MAINS_CLASS_FROM_NAME.keys(),
+        ),
+        vol.Required(CONF_CHARGER_TYPE): vol.In(
+            _CHARGER_CLASS_FROM_NAME.keys(),
+        ),
+        vol.Required(CONF_DEVELOPER_MODE, default=False): bool,
+    }
+)
+
 
 def get_charger(
     hass: HomeAssistant,
@@ -132,22 +145,9 @@ async def async_step_user(
             self.data = user_input
             return await self.async_step_mains()
 
-        schema = vol.Schema(
-            {
-                vol.Required(CONF_NAME): str,
-                vol.Required(CONF_MAINS_TYPE): vol.In(
-                    _MAINS_CLASS_FROM_NAME.keys(),
-                ),
-                vol.Required(CONF_CHARGER_TYPE): vol.In(
-                    _CHARGER_CLASS_FROM_NAME.keys(),
-                ),
-                vol.Required(CONF_DEVELOPER_MODE, default=False): bool,
-            }
-        )
-
         return self.async_show_form(
             step_id="user",
-            data_schema=schema,
+            data_schema=USER_SCHEMA,
             errors=errors,
         )
 

custom_components/ev_load_balancing/mains/virtual.py

@@ -0,0 +1,135 @@
+"""Handling Virtual mains currents input."""
+
+from datetime import UTC, datetime, timedelta
+import logging
+import statistics
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.event import async_track_state_change_event
+from homeassistant.helpers.template import device_entities
+
+from ..const import Phases
+from ..helpers.entity_value import get_sensor_entity_value
+from . import Mains, MainsPhase
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class MainsPhaseVirtual(MainsPhase):
+    """A data class for a mains phase."""
+
+    _stddev_min_num = 10
+    _stddev_max_age = timedelta(minutes=2)
+
+    def __init__(self, hass: HomeAssistant, entity_id: str) -> None:
+        """Initialize object."""
+        self._hass = hass
+        self._entity = entity_id
+        self._value = None
+        self._history_values = {}
+
+    def set_expected(self):
+        """Fake update mathod."""
+        pass
+
+    def update(self) -> None:
+        """Update measuremetns."""
+        now = datetime.now(UTC)
+        self._value = get_sensor_entity_value(
+            self._hass,
+            _LOGGER,
+            self._entity,
+        )
+
+        if self._value is None:
+            _LOGGER.debug("Skipping history since None value")
+            return
+
+        self._history_values[now] = self._value
+
+        # Find and drop old values if enough in dict
+        drop_keys = []
+        keep_count = 0
+        for k in sorted(self._history_values.keys(), reverse=True):
+            if keep_count < self._stddev_min_num or k > now - self._stddev_max_age:
+                keep_count += 1
+            else:
+                drop_keys.append(k)
+        for k in drop_keys:
+            self._history_values.pop(k)
+            _LOGGER.debug("Dropping measurement with key %s", k)
+
+    def actual_current(self) -> float:
+        """Get actual current on phase."""
+        return self._value
+
+    def stddev_current(self) -> float:
+        """Get standard deviation of current on phase."""
+        if len(self._history_values) > self._stddev_min_num / 2:
+            return statistics.pstdev(self._history_values.values())
+        _LOGGER.debug(
+            "Not enough values for stddev (%d), returning 0", len(self._history_values)
+        )
+        return 0
+
+
+class MainsVirtual(Mains):
+    """Virtual mains extractor."""
+
+    _state_change_listeners = []
+
+    def __init__(
+        self, hass: HomeAssistant, update_callback, device_id: str, mains_limit: int
+    ) -> None:
+        """Initilalize Virtual extractor."""
+        super().__init__(hass, update_callback)
+        self._id = device_id
+        self._mains_limit = mains_limit
+
+        entities = device_entities(hass, device_id)
+        used_entities = []
+
+        entity_phase1 = [e for e in entities if "_current" in e and e.endswith("1")][0]
+        self._phase1 = MainsPhaseVirtual(self._hass, entity_phase1)
+        used_entities.append(entity_phase1)
+
+        entity_phase2 = [e for e in entities if "_current" in e and e.endswith("2")][0]
+        self._phase2 = MainsPhaseVirtual(self._hass, entity_phase2)
+        used_entities.append(entity_phase2)
+
+        entity_phase3 = [e for e in entities if "_current" in e and e.endswith("3")][0]
+        self._phase3 = MainsPhaseVirtual(self._hass, entity_phase3)
+        used_entities.append(entity_phase3)
+
+        self._state_change_listeners.append(
+            async_track_state_change_event(
+                self._hass,
+                used_entities,
+                self._async_input_changed,
+            )
+        )
+
+    def get_phase(self, phase: Phases) -> MainsPhase:
+        """Return phase X data."""
+        if phase == Phases.PHASE1:
+            return self._phase1
+        if phase == Phases.PHASE2:
+            return self._phase2
+        if phase == Phases.PHASE3:
+            return self._phase3
+        return None
+
+    def get_rated_limit(self) -> int:
+        """Return main limit per phase."""
+        return self._mains_limit
+
+    def update(self) -> None:
+        """Update measuremetns."""
+        self._phase1.update()
+        self._phase2.update()
+        self._phase3.update()
+
+    def cleanup(self):
+        """Cleanup by removing event listeners."""
+        # for listner in self._state_change_listeners:
+        #     listner()

tests/test_config_flow.py

@@ -9,26 +9,28 @@
 # from pytest_homeassistant_custom_component.async_mock import patch
 import voluptuous as vol
 
-from homeassistant import config_entries
-from homeassistant.const import ATTR_NAME, ATTR_UNIT_OF_MEASUREMENT
-from homeassistant.helpers import selector
+# from homeassistant import config_entries
+# from homeassistant.const import ATTR_NAME, ATTR_UNIT_OF_MEASUREMENT
+# from homeassistant.helpers import selector
+from homeassistant.core import HomeAssistant
 
 
-# @pytest.mark.asyncio
-# async def test_flow_init(hass):
-#     """Test the initial flow."""
-#     result = await hass.config_entries.flow.async_init(
-#         config_flow.DOMAIN, context={"source": "user"}
-#     )
+@pytest.mark.asyncio
+async def test_flow_init(hass: HomeAssistant) -> None:
+    """Test the initial flow."""
+    result = await hass.config_entries.flow.async_init(
+        config_flow.DOMAIN, context={"source": "user"}
+    )
 
-#     expected = {
-#         "data_schema": SCHEMA_COPY,
-#         # "data_schema": config_flow.DATA_SCHEMA,
-#         "description_placeholders": None,
-#         "errors": {},
-#         "flow_id": mock.ANY,
-#         "handler": "ev_load_balancing",
-#         "step_id": "user",
-#         "type": "form",
-#     }
-#     assert expected == result
+    expected = {
+        "data_schema": config_flow.USER_SCHEMA,
+        "description_placeholders": None,
+        "errors": {},
+        "flow_id": mock.ANY,
+        "handler": "ev_load_balancing",
+        "last_step": None,
+        "preview": None,
+        "step_id": "user",
+        "type": "form",
+    }
+    assert expected == result