Merge pull request #47 from ie3-institute/to/#31-handle-appliance-ope…

…ration-state

Handle appliance operation state

.pre-commit-config.yaml

@@ -1,13 +1,13 @@
 repos:
--   repo: https://github.com/psf/black
-    rev: 22.6.0
-    hooks:
-    -   id: black
+- repo: https://github.com/psf/black
+  rev: 23.1.0
+  hooks:
+      - id: black
 - repo: https://github.com/PyCQA/flake8
-  rev: 5.0.4
+  rev: 6.0.0
   hooks:
       - id: flake8
 - repo: https://github.com/PyCQA/isort
-  rev: 5.10.1
+  rev: 5.12.0
   hooks:
       - id: isort

markovs_household/data/appliance.py

@@ -1,8 +1,9 @@
 import logging
-from abc import ABC
-from dataclasses import dataclass
-from datetime import datetime
-from typing import List
+import random
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta
+from typing import ClassVar, List
 
 from markovs_household.data.probability import (
     SwitchOnProbabilities,
@@ -35,6 +36,10 @@ def get_switch_on_probability(self, date_time: datetime) -> float:
             logging.error("Cannot determine the switch on probability", exc)
             raise exc
 
+    @abstractmethod
+    def get_operation_time(self) -> timedelta:
+        pass
+
 
 @dataclass(frozen=True)
 class ApplianceTypeLoadProfile(ApplianceType):
@@ -44,14 +49,21 @@ class ApplianceTypeLoadProfile(ApplianceType):
 
     profile: TimeSeries
 
+    def get_operation_time(self) -> timedelta:
+        return self.profile.length
+
 
 @dataclass(frozen=True)
 class ApplianceTypeConstantPower(ApplianceType):
     """
-    Appliance that has an associated constant power
+    Appliance that has an associated constant power and an operation time in seconds
     """
 
     power: float
+    operation_time: timedelta
+
+    def get_operation_time(self) -> timedelta:
+        return self.operation_time
 
 
 @dataclass(frozen=True)
@@ -61,4 +73,52 @@ class Appliance:
     """
 
     appliance_type: ApplianceType
-    operation_intervals: List[TimeInterval]
+    _operation_intervals: List[TimeInterval] = field(default_factory=list)
+    _random_generator: ClassVar[random.Random] = random.Random(42)
+
+    def handle_simulation_step(self, current_time: datetime) -> None:
+        """
+        Handles a simulation step to check stochastically check if the appliance is turned on at this point in time.
+        If the appliance is turned on an operation interval is added to the appliances operation intervals.
+        :param current_time: current time
+        """
+        if self.is_turned_on(current_time):
+            return
+        self._sample_switch_on(current_time)
+
+    def is_turned_on(self, current_time) -> bool:
+        if not self._operation_intervals:
+            return False
+        return self._operation_intervals[-1].is_within(current_time)
+
+    def _sample_switch_on(self, current_time) -> None:
+        """
+        Rolls the dice and compares it with the probability of the appliance to be turned on. If the dice roll falls
+        within the turn on probability of the device at the current time we "turn it on" by adding a corresponding
+        operation interval
+        :param current_time: current time
+        """
+        switch_on_probability_key = SwitchOnProbabilityKey.extract_from_datetime(
+            current_time
+        )
+        switch_on_probability = (
+            self.appliance_type.switch_on_probabilities.get_probability(
+                switch_on_probability_key
+            )
+        )
+        dice_roll = self._random_generator.random()
+        if dice_roll <= switch_on_probability:
+            self._add_operation_interval(current_time)
+
+    def _add_operation_interval(self, current_time: datetime):
+        self._operation_intervals.append(
+            TimeInterval.get_operation_interval(
+                current_time, self.appliance_type.get_operation_time()
+            )
+        )
+
+    def get_operation_intervals(self):
+        """
+        Returns the list of operation intervals of the appliance.
+        """
+        return self._operation_intervals

markovs_household/data/probability.py

@@ -37,7 +37,7 @@ class SwitchOnProbabilities:
     Probabilities to switch on an appliance given factors defined in the SwitchOnProbabilityKey class.
     """
 
-    __probabilities: Dict[SwitchOnProbabilityKey, float]
+    probabilities: Dict[SwitchOnProbabilityKey, float]
 
     @classmethod
     def from_csv(
@@ -86,7 +86,7 @@ def from_csv(
 
     def get_probability(self, key: SwitchOnProbabilityKey) -> float:
         try:
-            return self.__probabilities[key]
+            return self.probabilities[key]
         except KeyError as exc:
             logging.error(f"Couldn't find a switch on probability for key: {key}")
             raise exc

markovs_household/data/timeseries.py

@@ -1,13 +1,21 @@
 from dataclasses import dataclass
-from typing import Dict
+from datetime import timedelta
+
+
+@dataclass(frozen=True)
+class TimeSeriesEntry:
+    time: timedelta  # since start of time series
+    value: float  # in kW
 
 
 @dataclass(frozen=True)
 class TimeSeries:
     """
-    Data class to represent time series
-    :values: a mapping from a time step to its value
-    :type values: dict[int, float]
+    Data class to represent time series. The time series is configured in an event-discrete way. A value that is set
+    at a certain second value is valid until the start second value of the following entry.
+    :values: a mapping from a time in seconds (from operation start) to power in kW
+    :length: the total length of the time series
     """
 
-    values: Dict[int, float]
+    values: list[TimeSeriesEntry]
+    length: timedelta

markovs_household/data/usage_probabilities.py

@@ -8,7 +8,7 @@
 
 @dataclass(frozen=True)
 class UsageProbabilities:
-    __probabilities: Dict[str, float]
+    _probabilities: Dict[str, float]
 
     @classmethod
     def from_csv(cls, path: str) -> "UsageProbabilities":
@@ -27,7 +27,7 @@ def from_csv(cls, path: str) -> "UsageProbabilities":
         return UsageProbabilities(usage_probability_dict)
 
     def get_usage_probability(self, cat: ApplianceCategory):
-        return self.__probabilities[cat.value]
+        return self._probabilities[cat.value]
 
 
 usage_probability = UsageProbabilities

markovs_household/utils/time.py

@@ -75,11 +75,21 @@ class TimeInterval:
     end: datetime
 
     @classmethod
-    def get_operation_interval(cls, start: datetime, duration: timedelta):
+    def get_operation_interval(
+        cls, start: datetime, duration: timedelta
+    ) -> "TimeInterval":
         """
         Get the time interval from its start datetime and its duration
         :param start: start of the time interval
         :param duration: duration of the time interval
         :return:
         """
         return TimeInterval(start, start + duration)
+
+    def is_within(self, time: datetime) -> bool:
+        """
+        Checks the time step is within the time interval.
+        :param time: the time to check
+        :return: whether `time` lies within the interval
+        """
+        return True if (self.start <= time < self.end) else False

tests/common/test_data.py

@@ -1,12 +1,12 @@
 import random
-from datetime import datetime
+from datetime import datetime, timedelta
 
 from markovs_household.data.appliance import ApplianceTypeLoadProfile
 from markovs_household.data.probability import (
     SwitchOnProbabilities,
     SwitchOnProbabilityKey,
 )
-from markovs_household.data.timeseries import TimeSeries
+from markovs_household.data.timeseries import TimeSeries, TimeSeriesEntry
 from markovs_household.utils.appliance import ApplianceCategory
 from markovs_household.utils.time import DayType, Season
 
@@ -21,7 +21,14 @@
 RANDOM_SWITCH_ON_PROBABILITIES = SwitchOnProbabilities(
     {key: random.random() for key in SWITCH_ON_PROBABILITY_KEYS}
 )
-LOAD_PROFILE_STOVE = TimeSeries({quarterly_hour: 700 for quarterly_hour in range(4)})
+LOAD_PROFILE_STOVE = TimeSeries(
+    [
+        TimeSeriesEntry(timedelta(), 1),
+        TimeSeriesEntry(timedelta(seconds=60), 2),
+        TimeSeriesEntry(timedelta(seconds=120), 1),
+    ],
+    timedelta(minutes=4),
+)
 STOVE = ApplianceTypeLoadProfile(
     category=ApplianceCategory.STOVE,
     switch_on_probabilities=RANDOM_SWITCH_ON_PROBABILITIES,

tests/data/test_appliance.py

@@ -1,8 +1,16 @@
+from datetime import datetime, timedelta
+
 from markovs_household.data.appliance import (
+    Appliance,
+    ApplianceCategory,
     ApplianceTypeConstantPower,
     ApplianceTypeLoadProfile,
 )
-from markovs_household.utils.appliance import ApplianceCategory
+from markovs_household.data.probability import (
+    SwitchOnProbabilities,
+    SwitchOnProbabilityKey,
+)
+from markovs_household.utils.time import DayType, Season, TimeInterval
 from tests.common import test_data
 
 
@@ -27,8 +35,18 @@ def test_init_appliance_constant_profile():
         category=appliance_type,
         switch_on_probabilities=switch_on_probabilities,
         power=42,
+        operation_time=timedelta(minutes=10),
     )
-    assert appliance.category is ApplianceCategory.STOVE
+    assert appliance.category == ApplianceCategory.STOVE
+    assert appliance.switch_on_probabilities is test_data.RANDOM_SWITCH_ON_PROBABILITIES
+    assert appliance.power == 42
+    appliance = ApplianceTypeConstantPower(
+        category=appliance_type,
+        switch_on_probabilities=switch_on_probabilities,
+        power=42,
+        operation_time=timedelta(hours=1),
+    )
+    assert appliance.category == ApplianceCategory.STOVE
     assert appliance.switch_on_probabilities is test_data.RANDOM_SWITCH_ON_PROBABILITIES
     assert appliance.power == 42
 
@@ -38,3 +56,54 @@ def test_get_switch_on_probability():
     (datetime, key) = test_data.DATE_TIME_KEY_PAIR
     expected = stove.switch_on_probabilities.get_probability(key)
     assert stove.get_switch_on_probability(datetime) == expected
+    (dt, key) = test_data.DATE_TIME_KEY_PAIR
+    probabilities = stove.switch_on_probabilities.probabilities
+    expected = probabilities[SwitchOnProbabilityKey.extract_from_datetime(dt)]
+    assert stove.get_switch_on_probability(dt) == expected
+
+
+def test_is_turned_on():
+    appliance_type = ApplianceTypeLoadProfile(
+        category=ApplianceCategory.STOVE,
+        switch_on_probabilities=test_data.RANDOM_SWITCH_ON_PROBABILITIES,
+        profile=test_data.LOAD_PROFILE_STOVE,
+    )
+    operation_start = datetime(year=2021, month=11, day=11, hour=11, minute=11)
+    operation_end = operation_start + appliance_type.get_operation_time()
+    operation_interval = TimeInterval(operation_start, operation_end)
+    appliance = Appliance(
+        appliance_type=appliance_type, _operation_intervals=[operation_interval]
+    )
+    assert (
+        appliance.is_turned_on(
+            operation_start + appliance_type.get_operation_time() / 2
+        )
+        is True
+    )
+    assert (
+        appliance.is_turned_on(
+            operation_start + appliance_type.get_operation_time() + timedelta(seconds=1)
+        )
+        is False
+    )
+
+
+def test_handle_smiulation_step():
+    zero_probability_key = SwitchOnProbabilityKey(Season.WINTER, DayType.WEEKDAY, 0)
+    one_probability_key = SwitchOnProbabilityKey(Season.WINTER, DayType.WEEKDAY, 1)
+    switch_on_probabilities = SwitchOnProbabilities(
+        {zero_probability_key: 0, one_probability_key: 1}
+    )
+    appliance_type = ApplianceTypeConstantPower(
+        category=ApplianceCategory.STOVE,
+        switch_on_probabilities=switch_on_probabilities,
+        power=42,
+        operation_time=timedelta(minutes=10),
+    )
+    appliance = Appliance(appliance_type)
+    initial_time = datetime(year=2022, month=1, day=4, hour=0, minute=11)
+    appliance.handle_simulation_step(initial_time)
+    assert len(appliance.get_operation_intervals()) == 0
+    next_time = datetime(year=2022, month=1, day=4, hour=0, minute=24)
+    appliance.handle_simulation_step(next_time)
+    assert len(appliance.get_operation_intervals()) == 1

tests/input/test_household.py

@@ -1,3 +1,4 @@
+from datetime import timedelta
 from typing import Dict
 
 from markovs_household.data.appliance import (
@@ -22,6 +23,7 @@ class TestHouseholdAppliancesInput(HouseholdAppliancesInput):
             {SwitchOnProbabilityKey(Season.SPRING, DayType.WEEKDAY, 0): 0.1}
         ),
         50.0,
+        timedelta(hours=1),
     )
 
     video_recorder = ApplianceTypeConstantPower(
@@ -30,6 +32,7 @@ class TestHouseholdAppliancesInput(HouseholdAppliancesInput):
             {SwitchOnProbabilityKey(Season.SPRING, DayType.WEEKDAY, 0): 0.5}
         ),
         20.0,
+        timedelta(hours=1),
     )
 
     washing_machine = ApplianceTypeConstantPower(
@@ -38,6 +41,7 @@ class TestHouseholdAppliancesInput(HouseholdAppliancesInput):
             {SwitchOnProbabilityKey(Season.SPRING, DayType.WEEKDAY, 0): 0.5}
         ),
         100.0,
+        timedelta(hours=1),
     )
 
     @classmethod
@@ -99,7 +103,7 @@ def test_init_household_avg():
 
     for appliance in household.appliances:
         assert appliance.appliance_type == TestHouseholdAppliancesInput.pc
-        assert appliance.operation_intervals == []
+        assert appliance.get_operation_intervals() == []
 
 
 def test_init_household_by_no_of_inhabitants():
@@ -110,7 +114,7 @@ def test_init_household_by_no_of_inhabitants():
 
     for appliance in household.appliances:
         assert appliance.appliance_type == TestHouseholdAppliancesInput.pc
-        assert appliance.operation_intervals == []
+        assert appliance.get_operation_intervals() == []
 
 
 def test_init_household_by_income():