172 lines
6.3 KiB
Python
172 lines
6.3 KiB
Python
"""
|
|
Test FeatureExtractor và kiểm tra tích hợp với hệ thống
|
|
"""
|
|
|
|
import numpy as np
|
|
import xarray as xr
|
|
from feature_extractor import get_feature_extractor
|
|
from pathlib import Path
|
|
|
|
print("=" * 70)
|
|
print("TESTING FEATURE EXTRACTOR MODULE")
|
|
print("=" * 70)
|
|
|
|
# Test 1: Simple mode
|
|
print("\n[TEST 1] Simple Mode (3 features)")
|
|
print("-" * 50)
|
|
extractor_simple = get_feature_extractor(mode='simple')
|
|
print(f"✓ Created extractor: {extractor_simple.mode}")
|
|
print(f"✓ Expected features: {extractor_simple.config['n_features']}")
|
|
print(f"✓ Feature names: {extractor_simple.get_feature_names()}")
|
|
|
|
# Create dummy NDVI data
|
|
ndvi_dummy = xr.DataArray(
|
|
np.random.rand(10, 10),
|
|
dims=['y', 'x'],
|
|
coords={'y': np.arange(10), 'x': np.arange(10)}
|
|
)
|
|
vh_dummy = xr.DataArray(
|
|
np.random.rand(10, 10) * -10,
|
|
dims=['y', 'x'],
|
|
coords={'y': np.arange(10), 'x': np.arange(10)}
|
|
)
|
|
vv_dummy = xr.DataArray(
|
|
np.random.rand(10, 10) * -8,
|
|
dims=['y', 'x'],
|
|
coords={'y': np.arange(10), 'x': np.arange(10)}
|
|
)
|
|
|
|
features_simple = extractor_simple.extract(
|
|
ndvi_data=ndvi_dummy,
|
|
vh_data=vh_dummy,
|
|
vv_data=vv_dummy
|
|
)
|
|
print(f"✓ Extracted features shape: {features_simple.shape}")
|
|
assert features_simple.shape[1] == 3, "Expected 3 features"
|
|
print("✅ Simple mode test PASSED\n")
|
|
|
|
# Test 2: Extended mode
|
|
print("[TEST 2] Extended Mode (15 features)")
|
|
print("-" * 50)
|
|
extractor_extended = get_feature_extractor(mode='extended')
|
|
print(f"✓ Created extractor: {extractor_extended.mode}")
|
|
print(f"✓ Expected features: {extractor_extended.config['n_features']}")
|
|
print(f"✓ Feature names: {extractor_extended.get_feature_names()}")
|
|
|
|
# Create dummy S2 dataset with time dimension
|
|
s2_dummy = xr.Dataset({
|
|
'B02': xr.DataArray(np.random.rand(5, 10, 10), dims=['time', 'y', 'x']),
|
|
'B03': xr.DataArray(np.random.rand(5, 10, 10), dims=['time', 'y', 'x']),
|
|
'B04': xr.DataArray(np.random.rand(5, 10, 10), dims=['time', 'y', 'x']),
|
|
'B08': xr.DataArray(np.random.rand(5, 10, 10), dims=['time', 'y', 'x']),
|
|
'B11': xr.DataArray(np.random.rand(5, 10, 10), dims=['time', 'y', 'x'])
|
|
})
|
|
|
|
features_extended = extractor_extended.extract(
|
|
s2_data=s2_dummy,
|
|
vh_data=vh_dummy,
|
|
vv_data=vv_dummy
|
|
)
|
|
print(f"✓ Extracted features shape: {features_extended.shape}")
|
|
assert features_extended.shape[1] == 15, "Expected 15 features"
|
|
print("✅ Extended mode test PASSED\n")
|
|
|
|
# Test 3: Temporal mode
|
|
print("[TEST 3] Temporal Mode (39 features for 12 timesteps)")
|
|
print("-" * 50)
|
|
extractor_temporal = get_feature_extractor(mode='temporal')
|
|
print(f"✓ Created extractor: {extractor_temporal.mode}")
|
|
|
|
# Create dummy S2 dataset with 12 timesteps
|
|
s2_dummy_12 = xr.Dataset({
|
|
'B02': xr.DataArray(np.random.rand(12, 10, 10), dims=['time', 'y', 'x']),
|
|
'B03': xr.DataArray(np.random.rand(12, 10, 10), dims=['time', 'y', 'x']),
|
|
'B04': xr.DataArray(np.random.rand(12, 10, 10), dims=['time', 'y', 'x']),
|
|
'B08': xr.DataArray(np.random.rand(12, 10, 10), dims=['time', 'y', 'x']),
|
|
'B11': xr.DataArray(np.random.rand(12, 10, 10), dims=['time', 'y', 'x'])
|
|
})
|
|
|
|
features_temporal = extractor_temporal.extract(
|
|
s2_data=s2_dummy_12,
|
|
vh_data=vh_dummy,
|
|
vv_data=vv_dummy
|
|
)
|
|
|
|
# For temporal mode: 12 timesteps * 3 indices + 3 radar = 39 features
|
|
expected_features = 12 * 3 + 3
|
|
print(f"✓ Extracted features shape: {features_temporal.shape}")
|
|
print(f"✓ Expected: {expected_features} features (12 timesteps * 3 indices + 3 radar)")
|
|
|
|
feature_names_temporal = extractor_temporal.get_feature_names(n_timesteps=12)
|
|
print(f"✓ Feature names count: {len(feature_names_temporal)}")
|
|
print(f"✓ First 5 features: {feature_names_temporal[:5]}")
|
|
print(f"✓ Last 5 features: {feature_names_temporal[-5:]}")
|
|
|
|
assert features_temporal.shape[1] == expected_features, f"Expected {expected_features} features"
|
|
assert len(feature_names_temporal) == expected_features, f"Expected {expected_features} feature names"
|
|
print("✅ Temporal mode test PASSED\n")
|
|
|
|
# Test 4: Check model_odc.joblib metadata
|
|
print("[TEST 4] Verify model_odc.joblib metadata")
|
|
print("-" * 50)
|
|
metadata_file = Path("model_train/model_odc_info.json")
|
|
if metadata_file.exists():
|
|
import json
|
|
with open(metadata_file) as f:
|
|
metadata = json.load(f)
|
|
|
|
print(f"✓ Metadata file exists: {metadata_file}")
|
|
print(f"✓ Feature mode: {metadata.get('feature_mode')}")
|
|
print(f"✓ Number of features: {metadata.get('n_features')}")
|
|
print(f"✓ Features list length: {len(metadata.get('features', []))}")
|
|
print(f"✓ First 5 features: {metadata.get('features', [])[:5]}")
|
|
|
|
assert metadata.get('feature_mode') == 'temporal', "Expected temporal mode"
|
|
assert metadata.get('n_features') == 39, "Expected 39 features"
|
|
assert len(metadata.get('features', [])) == 39, "Expected 39 feature names"
|
|
|
|
print("✅ model_odc.joblib metadata VERIFIED\n")
|
|
else:
|
|
print("❌ model_odc_info.json not found. Run: python create_odc_metadata.py")
|
|
|
|
# Test 5: Check ModelManager integration
|
|
print("[TEST 5] Test ModelManager integration")
|
|
print("-" * 50)
|
|
try:
|
|
from model_manager import get_model_manager
|
|
|
|
manager = get_model_manager()
|
|
print(f"✓ ModelManager initialized")
|
|
|
|
# List models
|
|
models = manager.list_models()
|
|
print(f"✓ Found {len(models)} models")
|
|
|
|
# Check if model_odc.joblib has metadata
|
|
odc_model = next((m for m in models if m['filename'] == 'model_odc.joblib'), None)
|
|
if odc_model:
|
|
print(f"✓ model_odc.joblib found in list")
|
|
print(f" - Feature mode: {odc_model.get('feature_mode', 'N/A')}")
|
|
print(f" - N features: {odc_model.get('n_features', 'N/A')}")
|
|
print("✅ ModelManager integration test PASSED\n")
|
|
else:
|
|
print("⚠️ model_odc.joblib not in model list")
|
|
|
|
except Exception as e:
|
|
print(f"❌ ModelManager test failed: {e}")
|
|
|
|
# Summary
|
|
print("=" * 70)
|
|
print("TEST SUMMARY")
|
|
print("=" * 70)
|
|
print("✅ All feature extraction modes working correctly")
|
|
print("✅ Feature dimensions match expectations")
|
|
print("✅ Feature names generated correctly")
|
|
print("✅ model_odc.joblib metadata verified")
|
|
print("\nNext steps:")
|
|
print("1. Update api_server.py with run_prediction from run_prediction_new.py")
|
|
print("2. Test training with different feature_modes")
|
|
print("3. Test prediction with models using different modes")
|
|
print("\nSee UPDATE_SUMMARY.md for details.")
|
|
print("=" * 70)
|