Simulation loop

.vscode/launch.json

@@ -0,0 +1,15 @@
+{
+    // Use IntelliSense to learn about possible attributes.
+    // Hover to view descriptions of existing attributes.
+    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "Python: Main",
+            "type": "python",
+            "request": "launch",
+            "module": "markovs_household.main",
+            "justMyCode": true
+        }
+    ]
+}

markovs_household/config/config.py

@@ -1,4 +1,6 @@
 from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
 
 import yaml
 
@@ -7,16 +9,25 @@
 class Configuration:
     consider_load_shifting: bool
     simulate_typical_days: bool
-    switch_on_probabilities_datapath: str
-    usage_probabilities_file: str
+    start: datetime
+    end: datetime
+    probabilities_path: str
+    nr_houses: int
 
     @staticmethod
     def parse_config(path):
+        path = Path(path).resolve()
+
         with open(path, "r") as ymlfile:
             config = yaml.safe_load(ymlfile)
+        dt_fmt = "%Y-%m-%d %H:%M"
+        start = datetime.strptime(config["start"], dt_fmt)
+        end = datetime.strptime(config["end"], dt_fmt)
         return Configuration(
-            config["consider_load_shifting"],
-            config["simulate_typical_days"],
-            config["switch_on_probabilities_datapath"],
-            config["usage_probabilities_file"],
+            consider_load_shifting=config["consider_load_shifting"],
+            simulate_typical_days=config["simulate_typical_days"],
+            probabilities_path=config["probabilities_path"],
+            nr_houses=config["nr_houses"],
+            start=start,
+            end=end,
         )

markovs_household/data/appliance.py

@@ -60,7 +60,7 @@ class Appliance:
     _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:
+    def 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.

markovs_household/data/household.py

@@ -1,11 +1,15 @@
 import math
 import random
 from dataclasses import dataclass
+from datetime import datetime
 from typing import Dict, List
 
 from markovs_household.data.appliance import Appliance, ApplianceCategory, ApplianceType
 from markovs_household.data.household_categories import HouseholdIncome, HouseholdType
-from markovs_household.input.appliances_input import HouseholdAppliancesInput
+from markovs_household.input.appliances_input import (
+    ApplianceInitStrategy,
+    HouseholdAppliancesInput,
+)
 
 
 @dataclass(frozen=True)
@@ -17,6 +21,39 @@ class Household:
 
     appliances: List[Appliance]
 
+    def step(self, time: datetime):
+        [appliance.step(time) for appliance in self.appliances]
+
+    @classmethod
+    def initialize_households(
+        cls,
+        hh_input: HouseholdAppliancesInput,
+        strategy: ApplianceInitStrategy,
+        nr_houses: int,
+    ) -> list["Household"]:
+        # Initialize Households
+        households: list[Household] = []
+        match strategy:
+            case ApplianceInitStrategy.AVERAGE:
+                avg_appliances = hh_input.get_household_average_appliances()
+                for _ in range(nr_houses):
+                    appliances = []
+                    for cat in ApplianceCategory:
+                        cat_qty = avg_appliances[cat]
+                        nr, rmd = int(cat_qty), cat_qty % 1
+                        nr = int(nr)
+                        if rmd > random.random():
+                            nr += 1
+                        appliance_type = hh_input.get_appliance_types()[cat]
+                        [
+                            appliances.append(Appliance(appliance_type))
+                            for _ in range(nr)
+                        ]
+                    households.append(cls(appliances))
+            case _:
+                raise NotImplementedError(f"Strategy {strategy} not implemented yet")
+        return households
+
     @classmethod
     def from_average_household(cls, inp: HouseholdAppliancesInput) -> "Household":
         """

markovs_household/input/appliances_input.py

@@ -1,6 +1,7 @@
 import os.path
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
+from enum import Enum
 from typing import Dict
 
 import pandas as pd
@@ -19,6 +20,13 @@
 )
 
 
+class ApplianceInitStrategy(Enum):
+    AVERAGE = "average"
+    INCOME = "income"
+    INHABITANTS = "inhabitants"
+    TYPE = "type"
+
+
 class HouseholdAppliancesInput(ABC):
     @abstractmethod
     def get_appliance_types(self) -> Dict[ApplianceCategory, ApplianceType]:
@@ -65,7 +73,12 @@ def __init__(self, dir_path: str, delimiter: str):
         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]
+            val = average_hh[appliance.value]
+            if len(val) != 1:
+                raise ValueError(
+                    f"Appliance {appliance.value} has more than one or no values!"
+                )
+            average_hh_dict[appliance] = val.iloc[0]
         self.average_hh = average_hh_dict
 
         def get_category_dict(data: pd.DataFrame):