Merge pull request #111 from JW-Kraft/fix/duty-cycle-repetition

Fix/duty cycle repetition for productive use

CHANGELOG.md

@@ -4,8 +4,11 @@ Release History
 0.5.2 (dev)
 -----------
 
+**|new|**       Introduction of an Appliance parameter to model productive use duty cycles: `continuous_use_duty_cycle`
+
 0.5.1 (2024-02-08)
------------
+------------------
+
 **|fixed|**     Plotting a cloud of profiles from the command line is fixed
 
 0.5.0 (2023-12-06)

docs/source/input_parameters.rst

@@ -108,6 +108,13 @@ The **"allowed values"** column provide information about the format one should
      - integer
      - no
      - 0
+  * - continuous_use_duty_cycle
+     - NA
+     - {0,1}
+     - Duty cycle mode, 0 triggers once per switch-on event, 1 let the duty cycle repeat during the entire switch-on event
+     - integer
+     - no
+     - 1
    * - occasional_use
      - %
      - in [0,1]

ramp/core/constants.py

@@ -73,6 +73,7 @@ def switch_on_parameters():
     "func_cycle",
     "fixed",
     "fixed_cycle",
+    "continuous_duty_cycle",
     "occasional_use",
     "flat",
     "thermal_p_var",
@@ -115,6 +116,7 @@ def switch_on_parameters():
     "func_cycle",
     "fixed",
     "fixed_cycle",
+    "continuous_duty_cycle",
     "occasional_use",
     "flat",
     "thermal_p_var",

ramp/core/core.py

@@ -584,7 +584,10 @@ def load(self, filename: str) -> None:
                 :, ~user_df.columns.isin(["user_name", "num_users", "user_preference"])
             ].to_dict(orient="records"):
                 # assign Appliance arguments
-                appliance_parameters = {k: row[k] for k in APPLIANCE_ARGS}
+                appliance_parameters = {}
+                for k in APPLIANCE_ARGS:
+                    if k in row:
+                        appliance_parameters[k] = row[k]
 
                 # assign windows arguments
                 for k in WINDOWS_PARAMETERS:
@@ -966,6 +969,7 @@ def __init__(
         func_cycle: int = 1,
         fixed: str = "no",
         fixed_cycle: int = 0,
+        continuous_duty_cycle: int = 1,
         occasional_use: float = 1,
         flat: str = "no",
         thermal_p_var: int = 0,
@@ -998,6 +1002,13 @@ def __init__(
         func_cycle : int[0,1440], optional
             minimum time(minutes) the appliance is kept on after switch-on event, by default 1
 
+        continuous_duty_cycle : int[0,1], optional
+            if value is 0 : the duty cycle is executed once per switch on event (like a
+            welder, or other productive use appliances)
+            if value is 1 : whether the duty cycle are filling the whole switch on event of
+            the appliance (like a fridge or other continuous use appliance)
+            by default 1
+
         fixed : str, optional
             if 'yes', all the 'n' appliances of this kind are always switched-on together, by default "no"
 
@@ -1046,6 +1057,7 @@ def __init__(
         self.func_cycle = func_cycle
         self.fixed = fixed
         self.fixed_cycle = fixed_cycle
+        self.continuous_duty_cycle = continuous_duty_cycle
         self.occasional_use = occasional_use
         self.flat = flat
         self.thermal_p_var = thermal_p_var
@@ -1111,6 +1123,10 @@ def __init__(
         self.random_cycle2 = np.array([])
         self.random_cycle3 = np.array([])
 
+        # attribute used to know if a switch on event falls within a given duty cycle window
+        # if it is 0, then no switch on events happen within any duty cycle windows
+        self.current_duty_cycle_id = 0
+
     def save(self) -> pd.DataFrame:
         """returns a pd.DataFrame containing the appliance data
 
@@ -1420,18 +1436,19 @@ def update_daily_use(self, coincidence, power, indexes):
         if (
             self.fixed_cycle > 0
         ):  # evaluates if the app has some duty cycles to be considered
-            evaluate = np.round(np.mean(indexes)) if indexes.size > 0 else 0
-            # selects the proper duty cycle and puts the corresponding power values in the indexes range
-            if evaluate in range(self.cw11[0], self.cw11[1]) or evaluate in range(
-                self.cw12[0], self.cw12[1]
-            ):
+            # the proper duty cycle was selected in self.rand_switch_on_window()
+            # now setting the corresponding power values in the indexes range
+            if self.current_duty_cycle_id == 1:
                 np.put(self.daily_use, indexes, (self.random_cycle1 * coincidence))
-            elif evaluate in range(self.cw21[0], self.cw21[1]) or evaluate in range(
-                self.cw22[0], self.cw22[1]
-            ):
+            elif self.current_duty_cycle_id == 2:
                 np.put(self.daily_use, indexes, (self.random_cycle2 * coincidence))
-            else:
+            elif self.current_duty_cycle_id == 3:
                 np.put(self.daily_use, indexes, (self.random_cycle3 * coincidence))
+            else:
+                print(
+                    f"The app {self.name} has duty cycle option on, however the switch on event fell outside the provided duty cycle windows"
+                )
+
         else:  # if no duty cycles are specified, a regular switch_on event is modelled
             # randomises also the App Power if thermal_p_var is on
             np.put(
@@ -1735,6 +1752,46 @@ def rand_switch_on_window(self, rand_time: int):
                 raise ValueError(
                     "There is something fishy with upper limit in switch on..."
                 )
+
+            if (
+                self.fixed_cycle > 0
+            ):  # evaluates if the app has some duty cycles to be considered
+                evaluate = np.round(np.mean(indexes)) if indexes.size > 0 else 0
+                # selects the proper duty cycle
+                if (
+                    self.cw11[0] <= evaluate < self.cw11[1]
+                    or self.cw12[0] <= evaluate < self.cw12[1]
+                ):
+                    self.current_duty_cycle_id = 1
+                    duty_cycle_duration = len(self.random_cycle1)
+                elif (
+                    self.cw21[0] <= evaluate < self.cw21[1]
+                    or self.cw22[0] <= evaluate < self.cw22[1]
+                ):
+                    self.current_duty_cycle_id = 2
+                    duty_cycle_duration = len(self.random_cycle2)
+                elif (
+                    self.cw31[0] <= evaluate < self.cw31[1]
+                    or self.cw32[0] <= evaluate < self.cw32[1]
+                ):
+                    self.current_duty_cycle_id = 3
+                    duty_cycle_duration = len(self.random_cycle3)
+                else:
+                    print(
+                        f"The app {self.name} has duty cycle option on, however the switch on event fell outside the provided duty cycle windows"
+                    )
+                    # TODO previously duty_cycle3 was always considered as default if the evaluate proxy did neither
+                    #  get selected by duty_cycle1 nor duty_cycle2, for default is kept but not silently anymore in
+                    #  order to see wheather this is an issue or not
+                    self.current_duty_cycle_id = 3
+                    duty_cycle_duration = len(self.random_cycle3)
+
+                if (
+                    indexes.size > duty_cycle_duration
+                    and self.continuous_duty_cycle == 0
+                ):
+                    # Limit switch_on_window to duration of duty_cycle
+                    indexes = indexes[0:duty_cycle_duration]
         else:
             indexes = None
             # there are no available windows anymore

ramp/test/test_run.py

@@ -8,6 +8,8 @@
 # %% Import required modules
 import pandas as pd
 import matplotlib.pyplot as plt
+from ramp.core.core import UseCase
+from ramp.post_process import post_process as pp
 import os
 
 # %% Function to test the output against reference results
@@ -74,6 +76,7 @@ def series_to_average(profile_series, num_days):
     axes[n - 1].legend()
     axes[n - 1].set_xticklabels([])
     axes[n - 2].set_xticklabels([])
+    plt.show()
 
 
 # %% Testing the output and providing visual result
@@ -84,5 +87,25 @@ def series_to_average(profile_series, num_days):
 by the tested code changes. If any differences are there, the developers should 
 evaluate whether these are as expected/designed or not
 """
+from ramp.example.input_file_1 import User_list as User_list1
+from ramp.example.input_file_2 import User_list as User_list2
+from ramp.example.input_file_3 import User_list as User_list3
 
-test_output("../results", "../test", num_input_files=3)
+for i, ul in enumerate([User_list1, User_list2, User_list3]):
+    of_path = os.path.join(pp.BASE_PATH, "test", "results", f"output_file_{i + 1}.csv")
+    if os.path.exists(of_path) is False:
+        uc = UseCase(
+            users=ul,
+            parallel_processing=False,
+        )
+        uc.initialize(peak_enlarge=0.15, num_days=30)
+
+        Profiles_list = uc.generate_daily_load_profiles(flat=True)
+
+        pp.export_series(Profiles_list, ofname=of_path)
+
+test_output(
+    os.path.join(pp.BASE_PATH, "test", "results"),
+    os.path.join(pp.BASE_PATH, "test"),
+    num_input_files=3,
+)

tests/test_duty_cycle_repetition.py

@@ -0,0 +1,78 @@
+import os
+import pytest
+import numpy as np
+
+from ramp import User, UseCase
+import pandas as pd
+
+pd.options.plotting.backend = "plotly"
+import plotly.io as pio
+
+pio.renderers.default = "browser"
+
+
+@pytest.fixture
+def test_use_case():
+    # Create an instance of UseCase to test fix for the duty cycle repetition issue 78
+
+    # %% Create test user
+    test_user = User(user_name="test_user", num_users=1)
+
+    # Create test appliance
+    test_appliance = test_user.add_appliance(
+        name="test_appliance_with_duty_cycles",
+        func_time=4 * 60,  # runs for 2 hours per day
+        window_1=[6 * 60, 20 * 60],  # usage timeframe from 10:00 to 17:00
+        num_windows=1,
+        fixed_cycle=1,  # appliance uses duty cycles
+        # Duty cycle 1
+        p_11=8000,  # power of the first cycle
+        t_11=2,  # time needed for the first cycle
+        p_12=2000,  # power of the second cycle
+        t_12=18,  # time needed for the second cycle
+        continuous_duty_cycle=0,
+    )
+    # Create and initialize UseCase
+    duty_cycle_test_uc = UseCase(name="duty_cycle_test", users=[test_user])
+    duty_cycle_test_uc.initialize(num_days=3)
+
+    return duty_cycle_test_uc
+
+
+class TestUseCase:
+    @pytest.mark.usefixtures("test_use_case")
+    def test_daily_load_profile(self, test_use_case):
+        # Generate load profiles
+        daily_load_profile = pd.DataFrame(
+            test_use_case.generate_daily_load_profiles(),
+            index=test_use_case.datetimeindex,
+        )
+
+        # Count separate switch-on events -> whenever there is a jump in the load profile from smaller or equal 0.001
+        # to a value larger 0.001 (0.001 is used in the RAMP algorithm to "mark" available switch-on events, therefore
+        # load profiles are often not actually 0 even though there is no appliance use scheduled
+
+        # create a boolean mask for the condition
+        mask = (daily_load_profile[0] <= 0.001) & (
+            daily_load_profile[0].shift(-1) > 0.001
+        )
+        # Sum the number of switch on events
+        switch_on_count = mask.sum()
+
+        # Get the duration of the test duty cycle
+        test_duty_cycle_duration = (
+            test_use_case.appliances[0].t_11 + test_use_case.appliances[0].t_12
+        )
+
+        # Calculate how many switch-on events are expected
+        expected_switch_on_count = (
+            test_use_case.appliances[0].func_time / test_duty_cycle_duration
+        )
+
+        # In rare cases, it might happen that switch-on events are scheduled in direct succession
+        # Therefore, the test should not fail if the number of expected switch-on events is not fully reached.
+        # For the defined test, I allow 2 switch-on events less than expected
+
+        # The test fails, if there are 2 switch-on events less than expected
+
+        assert switch_on_count >= expected_switch_on_count - 2