Merge pull request #104

Parse household initialization data

.vscode/settings.json

@@ -0,0 +1,7 @@
+{
+  "python.testing.pytestArgs": [
+    "."
+  ],
+  "python.testing.unittestEnabled": false,
+  "python.testing.pytestEnabled": true
+}

markovs_household/data/appliance.py

@@ -5,10 +5,7 @@
 from datetime import datetime, timedelta
 from typing import ClassVar, List
 
-from markovs_household.data.probability import (
-    SwitchOnProbabilities,
-    SwitchOnProbabilityKey,
-)
+from markovs_household.data.probability import SwitchOnProbabilityKey
 from markovs_household.data.timeseries import TimeSeries
 from markovs_household.utils.appliance import ApplianceCategory
 from markovs_household.utils.time import TimeInterval
@@ -21,7 +18,7 @@ class ApplianceType(ABC):
     """
 
     category: ApplianceCategory
-    switch_on_probabilities: SwitchOnProbabilities
+    switch_on_probabilities: dict[SwitchOnProbabilityKey, float]
 
     def get_switch_on_probability(self, date_time: datetime) -> float:
         """
@@ -31,7 +28,7 @@ def get_switch_on_probability(self, date_time: datetime) -> float:
         """
         key = SwitchOnProbabilityKey.extract_from_datetime(date_time)
         try:
-            return self.switch_on_probabilities.get_probability(key)
+            return self.switch_on_probabilities[key]
         except KeyError as exc:
             logging.error("Cannot determine the switch on probability", exc)
             raise exc
@@ -53,19 +50,6 @@ def get_operation_time(self) -> timedelta:
         return self.profile.length
 
 
-@dataclass(frozen=True)
-class ApplianceTypeConstantPower(ApplianceType):
-    """
-    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)
 class Appliance:
     """
@@ -101,11 +85,9 @@ def _sample_switch_on(self, current_time) -> None:
         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
-            )
-        )
+        switch_on_probability = self.appliance_type.switch_on_probabilities[
+            switch_on_probability_key
+        ]
         dice_roll = self._random_generator.random()
         if dice_roll <= switch_on_probability:
             self._add_operation_interval(current_time)

markovs_household/data/probability.py

@@ -1,14 +1,7 @@
-import logging
-import os
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Dict
 
-import pandas as pd
-
-from markovs_household.data.usage_probabilities import UsageProbabilities
 from markovs_household.utils import time
-from markovs_household.utils.appliance import ApplianceCategory
 from markovs_household.utils.time import DayType, Season
 
 
@@ -30,63 +23,11 @@ def extract_from_datetime(date_time: datetime) -> "SwitchOnProbabilityKey":
         quarterly_hour_of_day = time.get_quarterly_hour_of_day(date_time)
         return SwitchOnProbabilityKey(season, day_type, quarterly_hour_of_day)
 
-
-@dataclass(frozen=True)
-class SwitchOnProbabilities:
-    """
-    Probabilities to switch on an appliance given factors defined in the SwitchOnProbabilityKey class.
-    """
-
-    probabilities: Dict[SwitchOnProbabilityKey, float]
-
-    @classmethod
-    def from_csv(
-        cls, cat: ApplianceCategory, path: str, usageprobs: UsageProbabilities
-    ) -> "SwitchOnProbabilities":
-        probabilities_df = pd.read_csv(
-            os.path.join(path + "/" + cat.value + ".csv"), sep=";"
-        )
-
-        # since the input values have an hourly resolution we repeat them 4 times to get values for all quarter hours
-        probabilities_df = probabilities_df.loc[
-            probabilities_df.index.repeat(4)
-        ].reset_index(drop=True)
-
-        probabilities_df["spring_weekday"] = (
-            probabilities_df["summer_weekday"] + probabilities_df["winter_weekday"]
-        ) / 2
-        probabilities_df["spring_saturday"] = (
-            probabilities_df["summer_saturday"] + probabilities_df["winter_saturday"]
-        ) / 2
-        probabilities_df["spring_sunday"] = (
-            probabilities_df["summer_sunday"] + probabilities_df["winter_sunday"]
-        ) / 2
-
-        probabilities_df["autumn_weekday"] = probabilities_df["spring_weekday"]
-        probabilities_df["autumn_saturday"] = probabilities_df["spring_saturday"]
-        probabilities_df["autumn_sunday"] = probabilities_df["spring_sunday"]
-
-        probabilities_df = probabilities_df / 4 * usageprobs.get_usage_probability(cat)
-
-        probability_keys = [
+    @staticmethod
+    def get_all():
+        return [
             SwitchOnProbabilityKey(season, day_type, quarterly_hour_of_day)
             for season in Season
             for day_type in DayType
             for quarterly_hour_of_day in range(4 * 24)
         ]
-
-        probabilities = {
-            probability_key: probabilities_df[
-                probability_key.season.value + "_" + probability_key.day_type.value
-            ][probability_key.quarterly_hour_of_day]
-            for probability_key in probability_keys
-        }
-
-        return cls(probabilities)
-
-    def get_probability(self, key: SwitchOnProbabilityKey) -> float:
-        try:
-            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,6 +1,8 @@
 from dataclasses import dataclass
 from datetime import timedelta
 
+from pandas import Series
+
 
 @dataclass(frozen=True)
 class TimeSeriesEntry:
@@ -19,3 +21,16 @@ class TimeSeries:
 
     values: list[TimeSeriesEntry]
     length: timedelta
+
+    @classmethod
+    def from_quarter_hour_series(cls, series: Series):
+        entries = []
+        for idx, x in enumerate(series.dropna()):
+            td = timedelta(minutes=idx)
+            entries.append(TimeSeriesEntry(td, x))
+        return TimeSeries(entries, timedelta(minutes=15 * len(series)))
+
+    @classmethod
+    def for_constant_running_load(cls, load: float):
+        # todo: check how we ensure that this load is always on
+        return TimeSeries([TimeSeriesEntry(timedelta(), load)], timedelta(hours=24))

markovs_household/input/appliances_input.py

@@ -1,38 +1,161 @@
+import os.path
 from abc import ABC, abstractmethod
+from dataclasses import dataclass
 from typing import Dict
 
-from markovs_household.data.appliance import ApplianceCategory, ApplianceType
+import pandas as pd
+
+from markovs_household.data.appliance import (
+    ApplianceCategory,
+    ApplianceType,
+    ApplianceTypeLoadProfile,
+)
 from markovs_household.data.household_categories import HouseholdIncome, HouseholdType
+from markovs_household.data.probability import SwitchOnProbabilityKey
+from markovs_household.data.timeseries import TimeSeries
+from markovs_household.input.probabilities import (
+    read_switch_on_probablities,
+    read_usage_probabilities,
+)
 
 
 class HouseholdAppliancesInput(ABC):
-    @classmethod
     @abstractmethod
-    def get_appliance_types(cls) -> Dict[ApplianceCategory, ApplianceType]:
+    def get_appliance_types(self) -> Dict[ApplianceCategory, ApplianceType]:
         pass
 
-    @classmethod
     @abstractmethod
-    def get_household_average_appliances(cls) -> Dict[ApplianceCategory, float]:
+    def get_household_average_appliances(self) -> Dict[ApplianceCategory, float]:
         pass
 
-    @classmethod
     @abstractmethod
     def get_household_average_appliances_by_no_of_inhabitants(
-        cls,
+        self,
     ) -> Dict[int, Dict[ApplianceCategory, float]]:
         pass
 
-    @classmethod
     @abstractmethod
     def get_household_average_appliances_by_income(
-        cls,
+        self,
     ) -> Dict[HouseholdIncome, Dict[ApplianceCategory, float]]:
         pass
 
-    @classmethod
     @abstractmethod
     def get_household_average_appliances_by_household_type(
-        cls,
+        self,
     ) -> Dict[HouseholdType, Dict[ApplianceCategory, float]]:
         pass
+
+
+@dataclass
+class CsvHouseholdAppliancesInput(HouseholdAppliancesInput):
+    average_hh: Dict[ApplianceCategory, float]
+    by_income: Dict[HouseholdIncome, Dict[ApplianceCategory, float]]
+    by_inhabitants: Dict[int, Dict[ApplianceCategory, float]]
+    by_type: Dict[HouseholdType, Dict[ApplianceCategory, float]]
+    appliance_types: Dict[ApplianceCategory, ApplianceType]
+
+    def __init__(self, dir_path: str, delimiter: str):
+        appliances_path = os.path.join(dir_path, "appliances")
+
+        average_hh = pd.read_csv(
+            os.path.join(appliances_path, "average_hh.csv"), delimiter=delimiter
+        )
+        average_hh_dict = {}
+        for appliance in ApplianceCategory:
+            if appliance.value not in average_hh.columns:
+                raise ValueError(f"Appliance {appliance.value} doesn't exist!")
+            average_hh_dict[appliance] = average_hh[appliance.value]
+        self.average_hh = average_hh_dict
+
+        def get_category_dict(data: pd.DataFrame):
+            res = {}
+            for idx, row in data.iterrows():
+                appliance_dict = {}
+                for appliance in ApplianceCategory:
+                    if appliance.value not in row.keys():
+                        raise ValueError(f"Appliance {appliance.value} doesn't exist!")
+                    appliance_dict[appliance] = row[appliance.value]
+                res[idx] = appliance_dict
+            return res
+
+        by_income = pd.read_csv(
+            os.path.join(appliances_path, "by_income.csv"), delimiter=delimiter
+        )
+        by_income["income"] = by_income["income"].apply(lambda x: HouseholdIncome(x))
+        by_income.set_index("income", inplace=True, drop=True)
+        by_income_dict = get_category_dict(by_income)
+        self.by_income = by_income_dict
+
+        by_inhabitants = pd.read_csv(
+            os.path.join(appliances_path, "by_inhabitants.csv"),
+            delimiter=delimiter,
+            index_col="inhabitants",
+        )
+        by_inhabitants_dict = get_category_dict(by_inhabitants)
+        self.by_inhabitants = by_inhabitants_dict
+
+        by_type = pd.read_csv(
+            os.path.join(appliances_path, "by_type.csv"), delimiter=delimiter
+        )
+        by_type["type"] = by_type["type"].apply(lambda x: HouseholdType(x))
+        by_type.set_index("type", inplace=True, drop=True)
+        by_type_dict = get_category_dict(by_type)
+        self.by_type = by_type_dict
+
+        self.appliance_types = self.initialize_appliance_types(
+            os.path.join(
+                dir_path,
+                "probabilities",
+                "usage_probabilities",
+                "usage_probabilities.csv",
+            ),
+            os.path.join(dir_path, "probabilities", "switch_on_probabilities"),
+            os.path.join(dir_path, "appliances", "load_ts.csv"),
+        )
+
+    @staticmethod
+    def initialize_appliance_types(
+        usage_probs_path: str, switch_on_probs_path: str, load_profile_path: str
+    ) -> dict[ApplianceCategory, ApplianceType]:
+        usage_probs = read_usage_probabilities(usage_probs_path)
+        load_profile_df = pd.read_csv(load_profile_path)
+        dct = {}
+        for cat in ApplianceCategory:
+            if cat == ApplianceCategory.OTHER_LOAD:
+                # We only expect a single value, as this is a constant load appliance
+                load = load_profile_df[cat.value][~load_profile_df[cat.value].isnull()]
+                assert len(load) == 1
+                load = load.iloc[0]
+                load_profile = TimeSeries.for_constant_running_load(load)
+                switch_on_probs = {key: 1.0 for key in SwitchOnProbabilityKey.get_all()}
+            else:
+                load_profile = TimeSeries.from_quarter_hour_series(
+                    load_profile_df[cat.value]
+                )
+                switch_on_probs = read_switch_on_probablities(
+                    cat, switch_on_probs_path, usage_probs[cat]
+                )
+            dct[cat] = ApplianceTypeLoadProfile(cat, switch_on_probs, load_profile)
+        return dct
+
+    def get_appliance_types(self) -> Dict[ApplianceCategory, ApplianceType]:
+        return self.appliance_types
+
+    def get_household_average_appliances(self) -> Dict[ApplianceCategory, float]:
+        return self.average_hh
+
+    def get_household_average_appliances_by_no_of_inhabitants(
+        self,
+    ) -> Dict[int, Dict[ApplianceCategory, float]]:
+        return self.by_inhabitants
+
+    def get_household_average_appliances_by_income(
+        self,
+    ) -> Dict[HouseholdIncome, Dict[ApplianceCategory, float]]:
+        return self.by_income
+
+    def get_household_average_appliances_by_household_type(
+        self,
+    ) -> Dict[HouseholdType, Dict[ApplianceCategory, float]]:
+        return self.by_type