Implement result handling

markovs_household/data/appliance.py

@@ -3,6 +3,7 @@
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta
+from math import ceil
 from typing import ClassVar, List
 
 from markovs_household.data.probability import (
@@ -40,6 +41,9 @@ def get_switch_on_probability(self, date_time: datetime) -> float:
     def get_operation_time(self) -> timedelta:
         pass
 
+    def get_timeseries_for(self, step_size: timedelta) -> List[float]:
+        pass
+
 
 @dataclass(frozen=True)
 class ApplianceTypeLoadProfile(ApplianceType):
@@ -52,6 +56,20 @@ class ApplianceTypeLoadProfile(ApplianceType):
     def get_operation_time(self) -> timedelta:
         return self.profile.length
 
+    def get_timeseries_for(self, step_size: timedelta) -> List[float]:
+        last = self.profile.values[0]
+
+        result = [last.value]
+
+        for entry in self.profile.values[1:]:
+            delta_seconds = entry.time - last.time
+
+            if delta_seconds == step_size.seconds:
+                result.append(entry.value)
+                last = entry
+
+        return result
+
 
 @dataclass(frozen=True)
 class ApplianceTypeConstantPower(ApplianceType):
@@ -65,6 +83,11 @@ class ApplianceTypeConstantPower(ApplianceType):
     def get_operation_time(self) -> timedelta:
         return self.operation_time
 
+    def get_timeseries_for(self, step_size: timedelta) -> List[float]:
+        steps = ceil(self.operation_time / step_size)
+
+        return [self.power] * steps
+
 
 @dataclass(frozen=True)
 class Appliance:
@@ -76,8 +99,7 @@ class Appliance:
     _operation_intervals: List[TimeInterval] = field(default_factory=list)
     random_generator: ClassVar[random.Random] = random.Random(42)
 
-    @property
-    def __operation_intervals(self):
+    def operation_intervals(self):
         return self._operation_intervals
 
     def handle_simulation_step(self, current_time: datetime) -> None:

markovs_household/output/result_sink.py

@@ -0,0 +1,37 @@
+import csv
+from datetime import timedelta, datetime
+
+from markovs_household.data.household import Household
+
+
+def create_timeseries(hh: Household, step_size: timedelta) -> dict[datetime, float]:
+    timeseries: dict[datetime, float] = {}
+
+    for appliance in hh.appliances:
+        intervals = appliance.operation_intervals()
+
+        for interval in intervals:
+            appliance_series = appliance.appliance_type.get_timeseries_for(step_size)
+            date = interval.start
+
+            for power in appliance_series:
+                old_value = timeseries.get(date, 0.0)
+                timeseries[date] = old_value + power
+
+                date = date + step_size
+
+            # set the value that follows the last of the appliance
+            # time series values to 0, if it has not been set yet
+            if timeseries.get(date) is None:
+                timeseries[date] = 0.0
+
+    return timeseries
+
+
+def write_timeseries(timeseries: dict[datetime, float], file: str):
+    with open(file, mode="w") as csv_file:
+        csv_writer = csv.writer(csv_file, delimiter=",")
+        csv_writer.writerow(["time", "power"])
+
+        for time, power in timeseries.items():
+            csv_writer.writerow([time, power])

tests/common/test_data.py

@@ -1,7 +1,10 @@
 import random
 from datetime import datetime, timedelta
 
-from markovs_household.data.appliance import ApplianceTypeLoadProfile
+from markovs_household.data.appliance import (
+    ApplianceTypeLoadProfile,
+    ApplianceTypeConstantPower,
+)
 from markovs_household.data.probability import (
     SwitchOnProbabilities,
     SwitchOnProbabilityKey,
@@ -18,18 +21,48 @@
 ]
 
 random.seed(42)
+
 RANDOM_SWITCH_ON_PROBABILITIES = SwitchOnProbabilities(
     {key: random.random() for key in SWITCH_ON_PROBABILITY_KEYS}
 )
 LOAD_PROFILE_STOVE = TimeSeries(
     [TimeSeriesEntry(0, 100), TimeSeriesEntry(60, 200), TimeSeriesEntry(120, 150)],
     timedelta(minutes=4),
 )
+
 STOVE = ApplianceTypeLoadProfile(
     category=ApplianceCategory.STOVE,
     switch_on_probabilities=RANDOM_SWITCH_ON_PROBABILITIES,
     profile=LOAD_PROFILE_STOVE,
 )
+
+PC = ApplianceTypeConstantPower(
+    ApplianceCategory.PC,
+    SwitchOnProbabilities(
+        {SwitchOnProbabilityKey(Season.SPRING, DayType.WEEKDAY, 0): 0.1}
+    ),
+    50.0,
+    timedelta(hours=1),
+)
+
+VIDEO_RECORDER = ApplianceTypeConstantPower(
+    ApplianceCategory.VIDEO_RECORDER,
+    SwitchOnProbabilities(
+        {SwitchOnProbabilityKey(Season.SPRING, DayType.WEEKDAY, 0): 0.5}
+    ),
+    20.0,
+    timedelta(hours=1),
+)
+
+WASHING_MACHINE = ApplianceTypeConstantPower(
+    ApplianceCategory.WASHING_MACHINE,
+    SwitchOnProbabilities(
+        {SwitchOnProbabilityKey(Season.SPRING, DayType.WEEKDAY, 0): 0.5}
+    ),
+    100.0,
+    timedelta(hours=1, minutes=30),
+)
+
 DATE_TIME_KEY_PAIR = (
     datetime(year=2021, month=11, day=16, hour=9, minute=0),
     SwitchOnProbabilityKey(Season.AUTUMN, DayType.WEEKDAY, 36),

tests/data/test_appliance.py

@@ -12,6 +12,7 @@
 )
 from markovs_household.utils.time import DayType, Season, TimeInterval
 from tests.common import test_data
+from tests.common.test_data import STOVE
 
 
 def test_init_appliance_load_profile():
@@ -52,19 +53,19 @@ def test_init_appliance_constant_profile():
 
 
 def test_get_switch_on_probability():
-    stove = test_data.STOVE
     (datetime, key) = test_data.DATE_TIME_KEY_PAIR
-    expected = stove.switch_on_probabilities.get_probability(key)
-    assert stove.get_switch_on_probability(datetime) == expected
+    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
+    probabilities = STOVE.switch_on_probabilities.probabilities
     expected = probabilities[SwitchOnProbabilityKey.extract_from_datetime(dt)]
-    assert stove.get_switch_on_probability(dt) == expected
+    assert STOVE.get_switch_on_probability(dt) == expected
 
 
 def test_is_turned_on():
     appliance_type = ApplianceTypeLoadProfile(
-        category=test_data.STOVE,
+        category=STOVE,
         switch_on_probabilities=test_data.RANDOM_SWITCH_ON_PROBABILITIES,
         profile=test_data.LOAD_PROFILE_STOVE,
     )
@@ -94,7 +95,7 @@ def test_handle_smiulation_step():
     switch_on_probabilities = SwitchOnProbabilities(
         {zero_probability_key: 0, one_probability_key: 1}
     )
-    stove = test_data.STOVE
+    stove = STOVE
     appliance_type = ApplianceTypeConstantPower(
         category=stove,
         switch_on_probabilities=switch_on_probabilities,

tests/input/test_household.py

@@ -1,55 +1,22 @@
-from datetime import timedelta
 from typing import Dict
 
 from markovs_household.data.appliance import (
     Appliance,
     ApplianceCategory,
     ApplianceType,
-    ApplianceTypeConstantPower,
 )
 from markovs_household.data.household import Household, HouseholdIncome, HouseholdType
-from markovs_household.data.probability import (
-    SwitchOnProbabilities,
-    SwitchOnProbabilityKey,
-)
 from markovs_household.input.appliances_input import HouseholdAppliancesInput
-from markovs_household.utils.time import DayType, Season
+from tests.common.test_data import PC, VIDEO_RECORDER, WASHING_MACHINE
 
 
 class TestHouseholdAppliancesInput(HouseholdAppliancesInput):
-    pc = ApplianceTypeConstantPower(
-        ApplianceCategory.PC,
-        SwitchOnProbabilities(
-            {SwitchOnProbabilityKey(Season.SPRING, DayType.WEEKDAY, 0): 0.1}
-        ),
-        50.0,
-        timedelta(hours=1),
-    )
-
-    video_recorder = ApplianceTypeConstantPower(
-        ApplianceCategory.VIDEO_RECORDER,
-        SwitchOnProbabilities(
-            {SwitchOnProbabilityKey(Season.SPRING, DayType.WEEKDAY, 0): 0.5}
-        ),
-        20.0,
-        timedelta(hours=1),
-    )
-
-    washing_machine = ApplianceTypeConstantPower(
-        ApplianceCategory.WASHING_MACHINE,
-        SwitchOnProbabilities(
-            {SwitchOnProbabilityKey(Season.SPRING, DayType.WEEKDAY, 0): 0.5}
-        ),
-        100.0,
-        timedelta(hours=1),
-    )
-
     @classmethod
     def get_appliance_types(cls) -> Dict[ApplianceCategory, ApplianceType]:
         return {
-            ApplianceCategory.PC: cls.pc,
-            ApplianceCategory.VIDEO_RECORDER: cls.video_recorder,
-            ApplianceCategory.WASHING_MACHINE: cls.washing_machine,
+            ApplianceCategory.PC: PC,
+            ApplianceCategory.VIDEO_RECORDER: VIDEO_RECORDER,
+            ApplianceCategory.WASHING_MACHINE: WASHING_MACHINE,
         }
 
     @classmethod
@@ -102,7 +69,7 @@ def test_init_household_avg():
     assert len(household.appliances) == 2
 
     for appliance in household.appliances:
-        assert appliance.appliance_type == TestHouseholdAppliancesInput.pc
+        assert appliance.appliance_type == PC
         assert appliance._operation_intervals == []
 
 
@@ -113,7 +80,7 @@ def test_init_household_by_no_of_inhabitants():
     assert len(household.appliances) == 2 or len(household.appliances) == 3
 
     for appliance in household.appliances:
-        assert appliance.appliance_type == TestHouseholdAppliancesInput.pc
+        assert appliance.appliance_type == PC
         assert appliance._operation_intervals == []
 
 
@@ -122,11 +89,8 @@ def test_init_household_by_income():
 
     assert len(household.appliances) == 3
 
-    assert (
-        Appliance(TestHouseholdAppliancesInput.washing_machine, [])
-        in household.appliances
-    )
-    assert Appliance(TestHouseholdAppliancesInput.pc, []) in household.appliances
+    assert Appliance(WASHING_MACHINE, []) in household.appliances
+    assert Appliance(PC, []) in household.appliances
 
 
 def test_init_household_by_household_type():
@@ -136,12 +100,6 @@ def test_init_household_by_household_type():
 
     assert len(household.appliances) == 5
 
-    assert (
-        Appliance(TestHouseholdAppliancesInput.washing_machine, [])
-        in household.appliances
-    )
-    assert Appliance(TestHouseholdAppliancesInput.pc, []) in household.appliances
-    assert (
-        Appliance(TestHouseholdAppliancesInput.video_recorder, [])
-        in household.appliances
-    )
+    assert Appliance(WASHING_MACHINE, []) in household.appliances
+    assert Appliance(PC, []) in household.appliances
+    assert Appliance(VIDEO_RECORDER, []) in household.appliances