Power

CPDShell/power.py

@@ -0,0 +1,39 @@
+from pathlib import Path
+
+from CPDShell.shell import CPContainer
+
+
+class Power:
+    def __init__(self, root_path: Path):
+        self.root_path = root_path
+
+    def __read_all_csv(self):
+        """Iterator for reading all CSV files in the specified directory and its subdirectories."""
+        for csv_file in self.root_path.rglob("*.csv"):
+            data = CPContainer.from_csv(csv_file)
+            yield data
+
+    @staticmethod
+    def __checking_near_point(container: CPContainer):
+        """Checks whether at least one change point is about the expected ones."""
+        margin = 50
+        detected_points = container.result
+        expected_points = container.expected_result
+
+        if not expected_points:
+            return False
+
+        for detected in detected_points:
+            if any(expected - margin <= detected <= expected + margin for expected in expected_points):
+                return True
+        return False
+
+    def calculate_power(self):
+        """Calculates the power of the algorithm based on hits about the disorder."""
+        false_negatives = 0
+        total_containers = sum(1 for _ in self.__read_all_csv())
+        for container in self.__read_all_csv():
+            if not self.__checking_near_point(container):
+                false_negatives += 1
+        probability_type_two_error = false_negatives / total_containers if total_containers > 0 else 0
+        return 1 - probability_type_two_error

CPDShell/shell.py

@@ -1,3 +1,4 @@
+import csv
 import time
 from collections.abc import Iterable, Sequence
 from pathlib import Path
@@ -90,6 +91,37 @@ def visualize(self, to_show: bool = True, output_directory: Path | None = None,
         if to_show:
             plt.show()
 
+    def to_csv(self, file_path: Path) -> None:
+        """Save the CPContainer data to a CSV file.
+
+        :param file_path: Path where the CSV file will be saved
+        """
+        # Ensure the directory exists
+        file_path.parent.mkdir(parents=True, exist_ok=True)
+
+        # Prepare the data for CSV
+        data_str = ",".join(map(str, self.data))
+        result_str = ",".join(map(str, self.result))
+        expected_result_str = ",".join(map(str, self.expected_result)) if self.expected_result else ""
+
+        # Write to CSV
+        with open(file_path, mode="w", newline="") as file:
+            writer = csv.writer(file)
+            writer.writerow(["data", "result", "expected_result", "time_sec"])
+            writer.writerow([data_str, result_str, expected_result_str, self.time_sec])
+
+    @classmethod
+    def from_csv(cls, file_path: Path) -> "CPContainer":
+        """Load CPContainer data from a CSV file."""
+        with open(file_path, newline="") as file:
+            reader = csv.DictReader(file)
+            row = next(reader)
+            data = list(map(float, row["data"].split(",")))
+            result = list(map(int, row["result"].split(",")))
+            expected_result = list(map(int, row["expected_result"].split(","))) if row["expected_result"] else None
+            time_sec = float(row["time_sec"])
+            return cls(data=data, result=result, expected_result=expected_result, time_sec=time_sec)
+
 
 class CPDShell:
     """Class, that grants a convenient interface to

tests/test_CPDShell/test_power.py

@@ -0,0 +1,91 @@
+import csv
+from pathlib import Path
+from tempfile import TemporaryDirectory
+
+import pytest
+
+from CPDShell.power import Power
+
+
+def create_mock_csv(file_path, data, result, expected_result, time_sec):
+    """Helper function to create a mock CSV file."""
+    with open(file_path, mode="w", newline="") as file:
+        writer = csv.writer(file)
+        writer.writerow(["data", "result", "expected_result", "time_sec"])
+        writer.writerow(
+            [
+                ",".join(map(str, data)),
+                ",".join(map(str, result)),
+                ",".join(map(str, expected_result)) if expected_result else "",
+                time_sec,
+            ]
+        )
+
+
+class TestPower:
+    @pytest.mark.parametrize(
+        "mock_files, expected_power, test_case_description",
+        [
+            (
+                [
+                    {"data": [1.0, 2.0, 3.0], "result": [10], "expected_result": [10], "time_sec": 0.1},
+                    {"data": [1.0, 2.0, 3.0], "result": [100], "expected_result": [10], "time_sec": 0.1},
+                ],
+                0.5,
+                "One correct detection, one false negative",
+            ),
+            (
+                [
+                    {"data": [1.0, 2.0, 3.0], "result": [10], "expected_result": [10], "time_sec": 0.1},
+                    {"data": [1.0, 2.0, 3.0], "result": [20], "expected_result": [20], "time_sec": 0.1},
+                ],
+                1.0,
+                "All correct detections",
+            ),
+            (
+                [
+                    {"data": [1.0, 2.0, 3.0], "result": [70], "expected_result": [0], "time_sec": 0.1},
+                    {"data": [1.0, 2.0, 3.0], "result": [80], "expected_result": [0], "time_sec": 0.1},
+                ],
+                0.0,
+                "All false negatives",
+            ),
+            (
+                [
+                    {"data": [1.0, 2.0, 3.0], "result": [1050], "expected_result": [1000], "time_sec": 0.1},
+                    {"data": [1.0, 2.0, 3.0], "result": [950], "expected_result": [1000], "time_sec": 0.1},
+                ],
+                1.0,
+                "Detections within margin range (50 units)",
+            ),
+            (
+                [{"data": [1.0, 2.0, 3.0], "result": [10], "expected_result": [], "time_sec": 0.1}],
+                0.0,
+                "No expected change points provided",
+            ),
+        ],
+    )
+    def test_calculate_power(self, mock_files, expected_power, test_case_description):
+        """Parameterized test for Power.calculate_power with various scenarios."""
+        with TemporaryDirectory() as temp_dir:
+            root_path = Path(temp_dir)
+
+            # Создание mock-файлов CSV
+            for i, mock_data in enumerate(mock_files):
+                create_mock_csv(
+                    root_path / f"test{i + 1}.csv",
+                    data=mock_data["data"],
+                    result=mock_data["result"],
+                    expected_result=mock_data["expected_result"],
+                    time_sec=mock_data["time_sec"],
+                )
+
+            # Инициализация Power и расчет мощности
+            power_calc = Power(root_path)
+            try:
+                power = power_calc.calculate_power()
+                assert power == pytest.approx(expected_power, 0.01), (
+                    f"Test case failed: {test_case_description}. " f"Expected power: {expected_power}, but got: {power}"
+                )
+            except TypeError as e:
+                pytest.fail(f"TypeError in test '{test_case_description}': {str(e)!s}")