diff --git a/fedot_ind/api/main.py b/fedot_ind/api/main.py index 0776a1c32..c07387219 100644 --- a/fedot_ind/api/main.py +++ b/fedot_ind/api/main.py @@ -241,11 +241,11 @@ def split_ts(self, time_series: np.array, strategy: str = 'frequent', plot: bool = True) -> Tuple[np.array, np.array]: - splitter = TSTransformer(time_series=time_series, - anomaly_dict=anomaly_dict, - strategy=strategy) + splitter = TSTransformer(strategy=strategy) - train_data, test_data = splitter.transform_for_fit(plot=plot, + train_data, test_data = splitter.transform_for_fit(series=time_series, + anomaly_dict=anomaly_dict, + plot=plot, binarize=binarize) return train_data, test_data diff --git a/tests/unit/api/test_api_main.py b/tests/unit/api/test_api_main.py index 35af56c04..33735035d 100644 --- a/tests/unit/api/test_api_main.py +++ b/tests/unit/api/test_api_main.py @@ -1,3 +1,6 @@ +import os.path + +import numpy as np import pytest from fedot_ind.api.main import FedotIndustrial @@ -5,57 +8,80 @@ from fedot_ind.core.architecture.experiment.TimeSeriesClassifier import TimeSeriesClassifier from fedot_ind.core.architecture.experiment.TimeSeriesClassifierPreset import TimeSeriesClassifierPreset from fedot_ind.core.models.topological.topological_extractor import TopologicalExtractor +from fedot_ind.tools.synthetic.ts_datasets_generator import TimeSeriesDatasetsGenerator @pytest.fixture() def tsc_topo_config(): - config = dict(task='ts_classification', - dataset='Chinatown', - strategy='topological', - timeout=0.5, - logging_level=40, - use_cache=False) - return config + return dict(task='ts_classification', + dataset='Chinatown', + strategy='topological', + timeout=0.1, + logging_level=40, + use_cache=False) @pytest.fixture() def tsc_fedot_preset_config(): - config = dict(task='ts_classification', - dataset='Chinatown', - strategy='fedot_preset', - timeout=0.5, - logging_level=40, - use_cache=False) - return config + return dict(task='ts_classification', + dataset='Chinatown', + strategy='fedot_preset', + timeout=0.5, + logging_level=40, + use_cache=False) @pytest.fixture() def none_tsc_config(): - config = dict(task='ts_classification', - dataset='Chinatown', - strategy=None, - timeout=0.5, - logging_level=40, - use_cache=False) - return config + return dict(task='ts_classification', + dataset='Chinatown', + strategy=None, + timeout=0.5, + logging_level=40, + use_cache=False) @pytest.fixture() def anomaly_detection_fedot_preset_config(): - config = dict(task='anomaly_detection', - dataset='custom_dataset', - strategy='fedot_preset', - use_cache=False, - timeout=0.5, - n_jobs=1, - logging_level=20) - return config + return dict(task='ts_forecasting', + dataset='custom_dataset', + strategy='fedot_preset', + use_cache=False, + timeout=0.5, + n_jobs=1, + logging_level=20) + + +@pytest.fixture() +def decomposition_config(): + return dict(task='anomaly_detection', + dataset='custom_dataset', + strategy='decomposition', + use_cache=False, + timeout=0.5, + n_jobs=1, + logging_level=20) + + +@pytest.fixture() +def ts_config(): + return dict(random_walk={'ts_type': 'random_walk', + 'length': 1000, + 'start_val': 36.6}) + + +@pytest.fixture() +def anomaly_config(): + return {'dip': {'level': 20, + 'number': 2, + 'min_anomaly_length': 10, + 'max_anomaly_length': 20} + } def test_main_api_topo(tsc_topo_config): industrial = FedotIndustrial(**tsc_topo_config) - assert type(industrial) is FedotIndustrial assert type(industrial.solver) is TimeSeriesClassifier assert industrial.solver.strategy == 'topological' assert industrial.config_dict['task'] == 'ts_classification' @@ -66,7 +92,6 @@ def test_main_api_topo(tsc_topo_config): def test_main_api_fedot_preset(tsc_fedot_preset_config): industrial = FedotIndustrial(**tsc_fedot_preset_config) - assert type(industrial) is FedotIndustrial assert type(industrial.solver) is TimeSeriesClassifierPreset assert industrial.solver.extractors == ['quantile_extractor', 'quantile_extractor', 'quantile_extractor'] assert industrial.solver.branch_nodes == ['eigen_basis', 'fourier_basis', 'wavelet_basis'] @@ -76,8 +101,64 @@ def test_main_api_fedot_preset(tsc_fedot_preset_config): def test_main_api_anomaly_detection_fedot_preset(anomaly_detection_fedot_preset_config): industrial = FedotIndustrial(**anomaly_detection_fedot_preset_config) - assert type(industrial) is FedotIndustrial + assert type(industrial.solver) is TimeSeriesAnomalyDetectionPreset assert industrial.solver.extractors == ['quantile_extractor', 'quantile_extractor', 'quantile_extractor'] assert industrial.solver.branch_nodes == ['eigen_basis', 'fourier_basis', 'wavelet_basis'] assert industrial.config_dict['task'] == 'anomaly_detection' + + +def test_api_tsc(tsc_topo_config): + tsc_topo_config.update({'output_folder': '.'}) + industrial = FedotIndustrial(**tsc_topo_config) + train_data, test_data = TimeSeriesDatasetsGenerator(num_samples=50, + max_ts_len=30, + n_classes=2, + test_size=0.5).generate_data() + model = industrial.fit(features=train_data[0], target=train_data[1]) + labels = industrial.predict(features=test_data[0], target=test_data[1]) + probs = industrial.predict_proba(features=test_data[0], target=test_data[1]) + metrics = industrial.get_metrics(target=test_data[1], metric_names=['roc_auc', 'accuracy']) + + for name, predict in zip(('labels', 'probs'), (labels, probs)): + industrial.save_predict(predicted_data=predict, kind=name) + industrial.save_metrics(metrics=metrics) + + expected_results_path = industrial.solver.saver.path + + for result in (model, labels, probs, metrics): + assert result is not None + for s in ('labels', 'probs', 'metrics'): + filepath = expected_results_path + f'/{s}.csv' + assert os.path.isfile(filepath) + + +def test_generate_ts(tsc_topo_config, ts_config): + industrial = FedotIndustrial(**tsc_topo_config) + ts = industrial.generate_ts(ts_config=ts_config) + + assert isinstance(ts, np.ndarray) + assert ts.shape[0] == 1000 + + +def test_generate_anomaly_ts(tsc_topo_config, ts_config, anomaly_config): + industrial = FedotIndustrial(**tsc_topo_config) + init_synth_ts, mod_synth_ts, synth_inters = industrial.generate_anomaly_ts(ts_data=ts_config, + anomaly_config=anomaly_config) + assert len(init_synth_ts) == len(mod_synth_ts) + for anomaly_type in synth_inters: + for interval in synth_inters[anomaly_type]: + ts_range = range(len(init_synth_ts)) + assert interval[0] in ts_range and interval[1] in ts_range + + +def test_split_ts(tsc_topo_config): + anomaly_dict = {'anomaly1': [[40, 50], [60, 80]], + 'anomaly2': [[130, 170], [300, 320]]} + industrial = FedotIndustrial(**tsc_topo_config) + train_data, test_data = industrial.split_ts(time_series=np.random.rand(1000), + anomaly_dict=anomaly_dict, + plot=False) + + assert train_data is not None + assert test_data is not None diff --git a/tests/unit/core/architecture/datasets/test_object_detection_datasets.py b/tests/unit/core/architecture/datasets/test_object_detection_datasets.py new file mode 100644 index 000000000..a6c0300bf --- /dev/null +++ b/tests/unit/core/architecture/datasets/test_object_detection_datasets.py @@ -0,0 +1,73 @@ +import json +from pathlib import Path + +import pytest +import yaml +from torchvision import transforms + +from fedot_ind.core.architecture.datasets.object_detection_datasets import COCODataset, YOLODataset + +synthetic_coco_data = { + "categories": [{"id": 1, "name": "cat"}, {"id": 2, "name": "dog"}], + "images": [ + {"id": 1, "file_name": "image1.jpg"}, + {"id": 2, "file_name": "image2.jpg"} + ], + "annotations": [ + {"image_id": 1, "category_id": 1, "area": 100, "bbox": [10, 20, 30, 40], "iscrowd": 0}, + {"image_id": 2, "category_id": 2, "area": 150, "bbox": [15, 25, 35, 45], "iscrowd": 1} + ] +} + + +synthetic_yolo_data = { + "train": "train/images", + "val": "val/images", + "names": ["cat", "dog"] +} + + +@pytest.fixture +def synthetic_coco_dataset(): + tmp_path = Path('.') + coco_json_path = tmp_path / "synthetic_coco.json" + coco_json_path.write_text(json.dumps(synthetic_coco_data)) + images_path = tmp_path / "images" + images_path.mkdir(exist_ok=True) + (images_path / "image1.jpg").write_text("") + (images_path / "image2.jpg").write_text("") + return COCODataset(str(images_path), str(coco_json_path), transform=transforms.ToTensor()) + + +@pytest.fixture +def synthetic_yolo_dataset(): + tmp_path = Path('.') + yolo_yaml_path = tmp_path / "synthetic_yolo.yaml" + yolo_yaml_path.write_text(yaml.dump(synthetic_yolo_data)) + root_path = tmp_path / "train" / "images" + root_path.mkdir(exist_ok=True, parents=True) + (root_path / "image1.jpg").write_text("") # Create empty files for images + (root_path / "image2.jpg").write_text("") + return YOLODataset(str(yolo_yaml_path), transform=transforms.ToTensor()) + + +def test_coco_dataset_length(synthetic_coco_dataset): + assert len(synthetic_coco_dataset) == 2 + + +def test_coco_dataset_sample(synthetic_coco_dataset): + sample = synthetic_coco_dataset.samples[0] + image, label = sample['image'], sample['labels'] + assert image is not None + assert label is not None + + +def test_yolo_dataset_length(synthetic_yolo_dataset): + assert len(synthetic_yolo_dataset) == 2 + + +def test_yolo_dataset_sample(synthetic_yolo_dataset): + sample = synthetic_yolo_dataset.samples[0] + image, label = sample + assert image is not None + assert label is not None diff --git a/tests/unit/core/operation/transformation/test_splitter.py b/tests/unit/core/operation/transformation/test_splitter.py index 9f1ef2cd5..d0478828b 100644 --- a/tests/unit/core/operation/transformation/test_splitter.py +++ b/tests/unit/core/operation/transformation/test_splitter.py @@ -17,7 +17,6 @@ def time_series(): return np.random.rand(320) -@pytest.fixture def anomaly_dict(): return {'anomaly1': [[40, 50], [60, 80]], 'anomaly2': [[130, 170], [300, 320]]}