#11 #20 set up custom pattern

elden1337 · Aug 14, 2023 · b6be681 · b6be681
1 parent 3dffe7c
commit b6be681
Show file tree

Hide file tree

Showing 11 changed files with 57 additions and 8 deletions.
diff --git a/custom_components/peaqnext/__init__.py b/custom_components/peaqnext/__init__.py
@@ -10,7 +10,7 @@
 from custom_components.peaqnext.service.models.consumption_type import ConsumptionType
 from custom_components.peaqnext.service.models.sensor_model import NextSensor
 
-from .const import (DOMAIN, PLATFORMS, HUB, CONF_NONHOURS_END, CONF_CONSUMPTION_TYPE, CONF_NAME, CONF_NONHOURS_START, CONF_SENSORS, CONF_TOTAL_CONSUMPTION_IN_KWH, CONF_TOTAL_DURATION_IN_MINUTES, CONF_CLOSEST_CHEAP)
+from .const import (CONF_CUSTOM_CONSUMPTION_PATTERN, DOMAIN, PLATFORMS, HUB, CONF_NONHOURS_END, CONF_CONSUMPTION_TYPE, CONF_NAME, CONF_NONHOURS_START, CONF_SENSORS, CONF_TOTAL_CONSUMPTION_IN_KWH, CONF_TOTAL_DURATION_IN_MINUTES, CONF_CLOSEST_CHEAP)
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -40,6 +40,7 @@ async def async_create_internal_sensors(conf: ConfigEntry) -> list[NextSensor]:
         sensors.append(
             NextSensor(
                 consumption_type=ConsumptionType(s[CONF_CONSUMPTION_TYPE]),
+                custom_consumption_pattern=s.get(CONF_CUSTOM_CONSUMPTION_PATTERN, None),
                 name=s[CONF_NAME],
                 hass_entity_id=nametoid(s[CONF_NAME]),
                 total_duration_in_seconds=s[CONF_TOTAL_DURATION_IN_MINUTES] * 60,

diff --git a/custom_components/peaqnext/config_flow.py b/custom_components/peaqnext/config_flow.py
@@ -21,7 +21,8 @@
     CONF_TOTAL_DURATION_IN_MINUTES,
     CONF_NONHOURS_START,
     CONF_NONHOURS_END,
-    CONF_CLOSEST_CHEAP
+    CONF_CLOSEST_CHEAP,
+    CONF_CUSTOM_CONSUMPTION_PATTERN
 )  # pylint:disable=unused-import
 
 _LOGGER = logging.getLogger(__name__)
@@ -31,6 +32,7 @@
     {
         vol.Required(CONF_NAME): cv.string,
         vol.Required(CONF_CONSUMPTION_TYPE): vol.In(CONSUMPTIONTYPE_NAMES),
+        vol.Optional(CONF_CUSTOM_CONSUMPTION_PATTERN): cv.string,
         vol.Required(CONF_TOTAL_CONSUMPTION_IN_KWH): cv.positive_float,
         vol.Required(CONF_TOTAL_DURATION_IN_MINUTES): cv.positive_float,
         vol.Optional(CONF_NONHOURS_START): cv.multi_select(list(range(0, 24))),
@@ -63,6 +65,7 @@ async def async_step_user(self, user_input=None, initial=True):
             self.data[CONF_SENSORS].append(
                 {
                     CONF_CONSUMPTION_TYPE: user_input[CONF_CONSUMPTION_TYPE],
+                    CONF_CUSTOM_CONSUMPTION_PATTERN: user_input.get(CONF_CUSTOM_CONSUMPTION_PATTERN, None),
                     CONF_NAME: user_input.get(CONF_NAME),
                     CONF_TOTAL_DURATION_IN_MINUTES: user_input.get(
                         CONF_TOTAL_DURATION_IN_MINUTES

diff --git a/custom_components/peaqnext/const.py b/custom_components/peaqnext/const.py
@@ -9,4 +9,5 @@
 CONF_TOTAL_CONSUMPTION_IN_KWH = "total_consumption_in_kwh"
 CONF_NONHOURS_START = "non_hours_start"
 CONF_NONHOURS_END = "non_hours_end"
-CONF_CLOSEST_CHEAP = "closest_cheap_hour"
+CONF_CLOSEST_CHEAP = "closest_cheap_hour"
+CONF_CUSTOM_CONSUMPTION_PATTERN = "custom_consumption_pattern"
diff --git a/custom_components/peaqnext/sensors/next_sensor.py b/custom_components/peaqnext/sensors/next_sensor.py
@@ -1,6 +1,8 @@
 """sensor implementation goes here"""
 from __future__ import annotations
 from typing import TYPE_CHECKING
+
+from custom_components.peaqnext.service.models.consumption_type import ConsumptionType
 from ..const import DOMAIN, HUB
 from datetime import datetime, timedelta
 
@@ -33,6 +35,7 @@ def __init__(self, hub: Hub, entry_id, given_name):
         self._non_hours_start = []
         self._non_hours_end = []
         self._closest_cheap_hour = None
+        self._custom_consumption_pattern = []
 
     @property
     def state(self) -> float:
@@ -53,6 +56,7 @@ async def async_update(self) -> None:
         self._non_hours_start = status.get("non_hours_start", [])
         self._non_hours_end = status.get("non_hours_end", [])
         self._closest_cheap_hour = status.get("closest_cheap_hour", 12)
+        self._custom_consumption_pattern = status.get("custom_consumption_pattern", [])
 
     @property
     def extra_state_attributes(self) -> dict:
@@ -67,6 +71,8 @@ def extra_state_attributes(self) -> dict:
             attr_dict["Non hours start"] = self._non_hours_start
         if len(self._non_hours_end) > 0:
             attr_dict["Non hours end"] = self._non_hours_end
+        if self._consumption_type == ConsumptionType.Custom.value:
+            attr_dict["Custom consumption pattern"] = self._custom_consumption_pattern
         attr_dict["raw_start"]= self._raw_start
         return attr_dict
 

diff --git a/custom_components/peaqnext/service/hub.py b/custom_components/peaqnext/service/hub.py
@@ -72,6 +72,7 @@ async def async_get_sensor_updates(self, active_sensor: NextSensor) -> dict:
             "non_hours_start": active_sensor.non_hours_start,
             "non_hours_end": active_sensor.non_hours_end,
             "closest_cheap_hour": active_sensor.default_closest_cheap,
+            "custom_consumption_pattern": active_sensor.custom_consumption_pattern_list,
         }
 
     @callback

diff --git a/custom_components/peaqnext/service/models/consumption_type.py b/custom_components/peaqnext/service/models/consumption_type.py
@@ -7,6 +7,7 @@ class ConsumptionType(Enum):
     PeakOut = "Peak in end"
     MidPeak = "Peak in middle"
     PeakInOut = "Peak in beginning and end"
+    Custom = "Custom consumption"
 
 
 CONSUMPTIONTYPE_NAMES = [e.value for e in ConsumptionType]
diff --git a/custom_components/peaqnext/service/models/sensor_model.py b/custom_components/peaqnext/service/models/sensor_model.py
@@ -18,11 +18,12 @@
 
 @dataclass
 class NextSensor:
-    consumption_type: ConsumptionType
+    consumption_type: ConsumptionType    
     name: str
     hass_entity_id: str
     total_duration_in_seconds: int
     total_consumption_in_kwh: float
+    custom_consumption_pattern: str|None = None
     default_closest_cheap: int = 12
     use_cent: bool = False
     non_hours_start: list[int] = field(default_factory=lambda: [])
@@ -31,6 +32,20 @@ class NextSensor:
     _mock_hour: int|None = None
     _mock_date: date|None = None
     _mock_minute: int|None = None
+    custom_consumption_pattern_list: list[int|float] = field(init=False)
+
+    def __post_init__(self):
+        self.custom_consumption_pattern_list = self._validate_custom_pattern(self.custom_consumption_pattern)
+
+    def _validate_custom_pattern(self, custom_consumption_pattern: str) -> list[int|float]:
+        if not custom_consumption_pattern:
+            raise Exception("No custom consumption pattern provided")
+        pattern = custom_consumption_pattern.split(",")
+        try:
+            pattern = [float(x) for x in pattern]
+        except Exception as e:
+            raise Exception("Invalid custom consumption pattern provided")
+        return pattern
 
     @property
     def best_start(self) -> HourModel:
@@ -77,6 +92,7 @@ def _update_sensor(self, prices: tuple[list,list], use_cent:bool = False, curren
             self.total_consumption_in_kwh,
             self.total_duration_in_seconds,
             self.consumption_type,
+            self.custom_consumption_pattern_list
         )
         try:            
             self._all_sequences = get_hours_sorted(

diff --git a/custom_components/peaqnext/service/segments.py b/custom_components/peaqnext/service/segments.py
@@ -9,18 +9,19 @@
     ConsumptionType.PeakOut: [1, 1, 1, 2],
     ConsumptionType.MidPeak: [1, 1, 2, 1, 1],
     ConsumptionType.PeakInOut: [2, 1, 1, 2],
+    ConsumptionType.Custom: []
 }
 
 _LOGGER = logging.getLogger(__name__)
 
 
 def calculate_consumption_per_hour(
-    consumption: float, duration_in_seconds: int, consumption_type: ConsumptionType
+    consumption: float, duration_in_seconds: int, consumption_type: ConsumptionType, custom_consumption_pattern: list
 ) -> list[float]:
     if 4 < duration_in_seconds <= 3600:
         return [consumption]
     try:
-        segments = _get_segments(consumption_type, duration_in_seconds)
+        segments = _get_segments(consumption_type, duration_in_seconds, custom_consumption_pattern)
     except Exception as e:
         _LOGGER.error(f"Unable to get segments for sensor. Exception: {e}")
         return [consumption]
@@ -57,9 +58,12 @@ def _get_minute_consumption(
 
 
 def _get_segments(
-    consumption_type: ConsumptionType, duration_in_seconds: int
+    consumption_type: ConsumptionType, duration_in_seconds: int, custom_consumption_pattern: list
 ) -> dict:
-    segments = CONSUMPTIONCONVERSION.get(consumption_type)
+    if consumption_type == ConsumptionType.Custom:
+        segments = custom_consumption_pattern
+    else:
+        segments = CONSUMPTIONCONVERSION.get(consumption_type)
     segment = 1
     ret = {}
     segment_duration = int(duration_in_seconds / len(segments) / 60) * 60
@@ -73,3 +77,5 @@ def _get_segments(
         r1 = ret[1][1] + missing
         ret[1] = (r0, r1)
     return ret
+
+
diff --git a/custom_components/peaqnext/strings.json b/custom_components/peaqnext/strings.json
@@ -6,6 +6,7 @@
                     "add_another_sensor": "Add another sensor?",
                     "name": "Name of the sensor.",
                     "consumption_type": "Pick consumption pattern",
+                    "custom_consumption_pattern": "(optional), add custom pattern as , separated list of values. e.g. 0.1,0.2 etc.",
                     "total_duration_in_minutes": "Cycle duration in minutes",
                     "total_consumption_in_kwh": "Expected consumption in kWh",
                     "non_hours_start": "Can't start in these hours",

diff --git a/custom_components/peaqnext/test/test_nextsensor.py b/custom_components/peaqnext/test/test_nextsensor.py
@@ -286,6 +286,18 @@ async def test_hub_this_hour_visible():
         #print(t.dt_start)
     assert 1 > 2
 
+@pytest.mark.asyncio
+async def test_custom_consumption_valid():    
+    s = NextSensor(consumption_type=ConsumptionType.Custom, name="test", hass_entity_id="sensor.test", total_duration_in_seconds=7200, total_consumption_in_kwh=10, custom_consumption_pattern="0.5,0.5") 
+    s.set_hour(4)   
+    await s.async_update_sensor((_p.P230729BE,[]))   
+    assert s.custom_consumption_pattern_list == [0.5,0.5]
+
+@pytest.mark.asyncio
+async def test_custom_consumption_invalid():    
+    with pytest.raises(Exception):   
+        s = NextSensor(consumption_type=ConsumptionType.Custom, name="test", hass_entity_id="sensor.test", total_duration_in_seconds=7200, total_consumption_in_kwh=10, custom_consumption_pattern="0.f5,0.5") 
+
 
 def _check_hourmodel(model: HourModel) -> bool:
     if model is None:

diff --git a/custom_components/peaqnext/translations/en.json b/custom_components/peaqnext/translations/en.json
@@ -6,6 +6,7 @@
                     "add_another_sensor": "Add another sensor?",
                     "name": "Name of the sensor.",
                     "consumption_type": "Pick consumption pattern",
+                    "custom_consumption_pattern": "(optional), add custom pattern as , separated list of values. e.g. 0.1,0.2 etc.",
                     "total_duration_in_minutes": "Cycle duration in minutes",
                     "total_consumption_in_kwh": "Expected consumption in kWh",
                     "non_hours_start": "Can't start in these hours",