Spaces:
Sleeping
Sleeping
File size: 5,561 Bytes
5841846 | 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 | import logging
import os
import time
from pathlib import Path
import httpx
import pandas as pd
logging.basicConfig(
level=logging.INFO,
format="[%(asctime)s] %(message)s",
datefmt="%H:%M:%S",
force=True
)
logging.getLogger("httpx").setLevel(logging.WARNING)
log = logging.getLogger(__name__)
class SmardFetcher:
BASE_URL = "https://www.smard.de/app/chart_data"
RESOLUTION = "hour"
START_DATE = pd.Timestamp("2020-01-01", tz="UTC")
def __init__(self, data_dir: str = None):
target_dir = data_dir or os.getenv("DATA_DIR", "data")
self.raw_dir = Path(target_dir) / "raw"
self.raw_dir.mkdir(parents=True, exist_ok=True)
self.dataset_file = self.raw_dir / "pvar_dataset.csv"
def _get_relevant_timestamps(self, client: httpx.Client, filter_id: str, region_id: str, last_dt: pd.Timestamp = None) -> list:
url = f"{self.BASE_URL}/{filter_id}/{region_id}/index_{self.RESOLUTION}.json"
response = client.get(url)
if response.status_code != 200:
return []
timestamps = response.json().get("timestamps", [])
start_ms = self.START_DATE.timestamp() * 1000
valid_ts = [ts for ts in timestamps if ts >= start_ms]
if last_dt:
last_ms = last_dt.timestamp() * 1000
valid_ts = [ts for ts in valid_ts if ts > last_ms]
return valid_ts
def _fetch_series(self, client: httpx.Client, filter_id: str, region_id: str, timestamp: int, col_name: str, retries=3) -> pd.DataFrame:
url = f"{self.BASE_URL}/{filter_id}/{region_id}/{filter_id}_{region_id}_{self.RESOLUTION}_{timestamp}.json"
for attempt in range(retries):
response = client.get(url)
if response.status_code == 200:
data = response.json().get("series", [])
if not data:
return pd.DataFrame()
df = pd.DataFrame(data, columns=["timestamp", col_name])
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
return df.set_index("timestamp")
elif response.status_code == 429:
time.sleep(2)
else:
break
return pd.DataFrame()
def _sync_feature(self, client: httpx.Client, filter_id: str, region_id: str, col_name: str, last_dt: pd.Timestamp = None) -> pd.DataFrame:
timestamps = self._get_relevant_timestamps(client, filter_id, region_id, last_dt)
if not timestamps:
return pd.DataFrame()
log.info(f" -> {col_name:<18} ({region_id}) | Downloading {len(timestamps)} blocks...")
dfs = []
for ts in timestamps:
dfs.append(self._fetch_series(client, filter_id, region_id, ts, col_name))
time.sleep(0.02) # Rate limit safety
valid_dfs = [df for df in dfs if not df.empty]
return pd.concat(valid_dfs).sort_index() if valid_dfs else pd.DataFrame()
def sync_all(self, progress_cb=None) -> pd.DataFrame:
# Hard wipe of the corrupted file to start 100% fresh
if self.dataset_file.exists():
log.warning("Wiping corrupted dataset to build a seamless combined version...")
self.dataset_file.unlink(missing_ok=True)
feature_dfs = []
with httpx.Client(timeout=30.0) as client:
# 1. Fetch Era 1 & 2 Prices
if progress_cb:
progress_cb("Lade Strompreise herunter...")
df_price_legacy = self._sync_feature(client, "4169", "DE-LU", "price_mwh")
df_price_modern = self._sync_feature(client, "4169", "DE", "price_mwh")
# Combine both price eras vertically, drop overlapping duplicates
df_price = pd.concat([df_price_legacy, df_price_modern]).sort_index()
df_price = df_price[~df_price.index.duplicated(keep='first')]
if not df_price.empty:
feature_dfs.append(df_price)
# 3. Fetch grid physics (Always under DE)
grid_filters = {
"410": "load_total",
"125": "prog_pv",
"123": "prog_wind_onshore",
"3791": "prog_wind_offshore"
}
grid_names = {
"load_total": "Netzlast",
"prog_pv": "Solarprognose",
"prog_wind_onshore": "Wind Onshore Prognose",
"prog_wind_offshore": "Wind Offshore Prognose"
}
for fid, col in grid_filters.items():
if progress_cb:
progress_cb(f"Lade {grid_names.get(col, col)} herunter...")
df_feat = self._sync_feature(client, fid, "DE", col)
if not df_feat.empty:
feature_dfs.append(df_feat)
if not feature_dfs:
log.error("No data could be retrieved.")
return pd.DataFrame()
if progress_cb:
progress_cb("Verarbeite Daten...")
# Merge all features side-by-side on the timestamp index
df_final = pd.concat(feature_dfs, axis=1).sort_index()
# Keep only data from 2020 onwards
df_final = df_final[df_final.index >= self.START_DATE]
df_final.to_csv(self.dataset_file)
log.info(f"Successfully generated clean dataset: {self.dataset_file} ({len(df_final)} rows)")
return df_final
if __name__ == "__main__":
fetcher = SmardFetcher()
fetcher.sync_all() |