diff --git a/CPDShell/power.py b/CPDShell/power.py new file mode 100644 index 0000000..d09d328 --- /dev/null +++ b/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 diff --git a/CPDShell/shell.py b/CPDShell/shell.py index 4431a0b..d780f86 100644 --- a/CPDShell/shell.py +++ b/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 diff --git a/tests/test_CPDShell/test_power.py b/tests/test_CPDShell/test_power.py new file mode 100644 index 0000000..374e0f1 --- /dev/null +++ b/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}")