Spaces:
Sleeping
Sleeping
File size: 5,876 Bytes
27cb60a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 |
"""
Phase 1 FOLLOW-UP: Detailed Investigation
==========================================
Investigates specific issues from initial tests:
1. Check 'businesstype' column (lowercase) for A53/A54
2. Find correct EIC column for CNEC filtering
3. Investigate pumping consumption query method
"""
import os
import pandas as pd
import polars as pl
from dotenv import load_dotenv
from entsoe import EntsoePandasClient
from pathlib import Path
load_dotenv()
API_KEY = os.getenv('ENTSOE_API_KEY')
client = EntsoePandasClient(api_key=API_KEY)
print("="*80)
print("PHASE 1 DETAILED INVESTIGATION")
print("="*80)
print()
# ============================================================================
# Investigation 1: businesstype column (lowercase)
# ============================================================================
print("-"*80)
print("INVESTIGATION 1: businesstype column analysis")
print("-"*80)
print()
try:
test_outages = client.query_unavailability_transmission(
country_code_from='10Y1001A1001A82H', # DE_LU
country_code_to='10YFR-RTE------C', # FR
start=pd.Timestamp('2025-09-23', tz='UTC'),
end=pd.Timestamp('2025-09-30', tz='UTC')
)
print(f"Outages returned: {len(test_outages)}")
print(f"\nAll columns:")
for i, col in enumerate(test_outages.columns, 1):
print(f" {i}. {col}")
print()
# Check lowercase businesstype
if 'businesstype' in test_outages.columns:
print("[OK] Found 'businesstype' column (lowercase)")
print("\nBusiness types distribution:")
business_counts = test_outages['businesstype'].value_counts()
print(business_counts)
print()
# Check for A53/A54
has_a53 = any('A53' in str(x) for x in test_outages['businesstype'].unique())
has_a54 = any('A54' in str(x) for x in test_outages['businesstype'].unique())
print(f"Contains A53 (Planned): {has_a53}")
print(f"Contains A54 (Unplanned): {has_a54}")
print()
# Show sample values
print("Sample businesstype values:")
print(test_outages['businesstype'].unique()[:10])
else:
print("[FAIL] businesstype column not found")
print()
# ========================================================================
# Investigation 2: Find CNEC/transmission element EIC column
# ========================================================================
print("-"*80)
print("INVESTIGATION 2: Finding transmission element EIC codes")
print("-"*80)
print()
print("Searching for columns containing 'eic', 'mrid', 'resource', 'asset', 'line'...")
print()
potential_cols = [col for col in test_outages.columns
if any(keyword in col.lower() for keyword in ['eic', 'mrid', 'resource', 'asset', 'line', 'domain'])]
print(f"Potential EIC columns: {potential_cols}")
print()
for col in potential_cols:
print(f"Column: {col}")
print(f" Sample values: {test_outages[col].unique()[:5].tolist()}")
print(f" Unique count: {test_outages[col].nunique()}")
print()
# Show full first record
print("Full first record:")
print(test_outages.iloc[0])
except Exception as e:
print(f"[FAIL] Investigation failed: {e}")
print()
# ============================================================================
# Investigation 3: Pumping consumption query methods
# ============================================================================
print("-"*80)
print("INVESTIGATION 3: Pumping consumption query options")
print("-"*80)
print()
print("Testing if pumping consumption is available via different queries...")
print()
# Try query_load (might include pumped storage consumption)
print("Option 1: Check if query_load() includes pumped storage consumption")
try:
load_ch = client.query_load(
country_code='CH',
start=pd.Timestamp('2025-09-23', tz='UTC'),
end=pd.Timestamp('2025-09-24', tz='UTC')
)
print(f"[OK] query_load() successful")
print(f" Type: {type(load_ch)}")
if isinstance(load_ch, pd.DataFrame):
print(f" Columns: {load_ch.columns.tolist()}")
print(f" Sample: {load_ch.head()}")
except Exception as e:
print(f"[FAIL] query_load() failed: {e}")
print()
# Try different PSR types
print("Option 2: Try different PSR types for pumped storage")
print(" PSR B10: Hydro Pumped Storage")
print(" PSR B11: Hydro Water Reservoir")
print(" PSR B12: Hydro Run-of-river")
print()
try:
# B10 already tested - get it again
gen_b10 = client.query_generation(
country_code='CH',
start=pd.Timestamp('2025-09-23 00:00', tz='UTC'),
end=pd.Timestamp('2025-09-23 23:00', tz='UTC'),
psr_type='B10'
)
print("[OK] PSR B10 (Pumped Storage) - Already tested")
print(f" Min: {gen_b10.min().values[0]:.2f} MW")
print(f" Max: {gen_b10.max().values[0]:.2f} MW")
print(f" Negative values: {(gen_b10 < 0).sum().values[0]}")
print()
# Check if there's a separate consumption metric
print("Checking entsoe-py methods for pumped storage consumption...")
print("Available methods:")
methods = [m for m in dir(client) if 'pump' in m.lower() or 'stor' in m.lower() or 'consum' in m.lower()]
if methods:
for method in methods:
print(f" - {method}")
else:
print(" >> No methods found with 'pump', 'stor', or 'consum' in name")
except Exception as e:
print(f"[FAIL] PSR type investigation failed: {e}")
print()
print("="*80)
print("INVESTIGATION COMPLETE")
print("="*80)
print()
print("Next Steps:")
print("1. Verify businesstype column contains A53/A54")
print("2. Identify correct EIC column for CNEC filtering")
print("3. Determine if pumping consumption is available (may need to infer from load data)")
print("="*80)
|