units for detection datasets

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

tests/unit/api/test_api_main.py

@@ -1,61 +1,87 @@
+import os.path
+
+import numpy as np
 import pytest
 
 from fedot_ind.api.main import FedotIndustrial
 from fedot_ind.core.architecture.experiment.TimeSeriesAnomalyDetection import TimeSeriesAnomalyDetectionPreset
 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

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

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]]}