Merge pull request #24 from gitaroktato/utility-functions

feat: Using utility functions
gitaroktato · Apr 10, 2024 · 0051690 · 0051690
2 parents d323de5 + 55464b1
commit 0051690
Show file tree

Hide file tree

Showing 5 changed files with 104 additions and 26 deletions.
diff --git a/bursty-traffic.ipynb b/bursty-traffic.ipynb
diff --git a/interesting-experiments.ipynb b/interesting-experiments.ipynb
diff --git a/simulating-md1-queues.ipynb b/simulating-md1-queues.ipynb
diff --git a/src/queue.py b/src/queue.py
@@ -1,6 +1,23 @@
 import numpy as np
 import numpy.typing as npt
 
+
+def intervals_to_timestamps(intervals: npt.NDArray[np.float64]) -> npt.NDArray[np.float64]:
+    timestamps = np.copy(intervals)
+    for index, value in enumerate(timestamps):
+        if index != 0:
+            timestamps[index] += timestamps[index - 1]
+    return timestamps
+
+
+def timestamps_to_intervals(timestamps: npt.NDArray[np.float64]) -> npt.NDArray[np.float64]:
+    intervals = np.copy(timestamps)
+    for index, value in reversed(list(enumerate(intervals))):
+        if index != 0:
+            intervals[index] -= intervals[index - 1]
+    return intervals
+
+
 class Queue:
 
     def __init__(
@@ -26,10 +43,7 @@ def process(self) -> None:
         self.__process_length()
 
     def __process_arrival_times(self) -> None:
-        self.__arrival_times = np.copy(self.__inter_arrival_times)
-        for index, value in enumerate(self.__arrival_times):
-            if index != 0:
-                self.__arrival_times[index] += self.__arrival_times[index-1]
+        self.__arrival_times = intervals_to_timestamps(self.__inter_arrival_times)
 
     def __process_departure_times(self) -> None:
         for index, arrive_at in enumerate(self.__arrival_times):

diff --git a/test/queue_test.py b/test/queue_test.py
@@ -1,6 +1,6 @@
 import numpy as np
 
-from src.queue import Queue
+from src.queue import Queue, intervals_to_timestamps, timestamps_to_intervals
 
 
 class TestQueue:
@@ -69,3 +69,13 @@ def test_with_two_executors(self):
         np.testing.assert_allclose(queue.utilization(1), [1, 1, 1, 1], rtol=1e-02)
 
 
+class TestIntervalsAndTimestamps:
+
+    def test_intervals_to_timestamps(self):
+        result = intervals_to_timestamps(np.array([1, 1, 1, 1]))
+        np.testing.assert_equal(result, [1, 2, 3, 4])
+
+    def test_timestamps_to_intervals(self):
+        result = timestamps_to_intervals(np.array([1, 2, 3, 4]))
+        np.testing.assert_equal(result, [1, 1, 1, 1])
+