File size: 4,368 Bytes
938949f b3e1b1d 938949f b3e1b1d 938949f 6d7f91e 938949f b3e1b1d 938949f b3e1b1d 938949f | 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 | """
Day-ahead planner worker.
Entry point for GitHub Actions cron (daily 05:00 IST = 02:00 UTC).
Usage:
python -m backend.workers.daily_planner
"""
from __future__ import annotations
import json
import logging
import sys
from datetime import date, datetime, timezone
from pathlib import Path
# Ensure project root is on sys.path
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
# Load .env if present (local dev)
try:
from dotenv import load_dotenv
load_dotenv(PROJECT_ROOT / ".env")
except ImportError:
pass
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
log = logging.getLogger("daily_planner")
def _get_forecast(target: date):
"""Build 96-slot (15-min) forecast arrays from IMS cache or defaults."""
try:
from src.data.ims_client import IMSClient
import pandas as pd
client = IMSClient()
df = client.load_cached()
if df.empty:
raise ValueError("No IMS cache available")
# Get the most recent day's pattern as a proxy for tomorrow
df["hour"] = pd.to_datetime(df["timestamp_utc"]).dt.hour
# Build 96 slots of temp and GHI
temps = []
ghis = []
for slot in range(96):
h = slot // 4
# Use mean values per hour from recent data
hour_data = df[df["hour"] == h]
t = hour_data["air_temperature_c"].mean() if "air_temperature_c" in df.columns and not hour_data.empty else 25.0
g = hour_data["ghi_w_m2"].mean() if "ghi_w_m2" in df.columns and not hour_data.empty else 0.0
temps.append(float(t) if not pd.isna(t) else 25.0)
ghis.append(float(g) if not pd.isna(g) else 0.0)
return temps, ghis
except Exception as exc:
log.warning("Could not build forecast from IMS: %s — using defaults", exc)
# Default: sinusoidal temperature and GHI profile for Sde Boker
import math
temps = []
ghis = []
for slot in range(96):
h = slot / 4 # fractional hour
# Temp: 18°C at night, peaks ~35°C at 14:00
t = 26.5 + 8.5 * math.sin(math.pi * (h - 6) / 12) if 6 <= h <= 18 else 20.0
# GHI: 0 at night, peaks ~900 W/m² at noon
g = max(0, 900 * math.sin(math.pi * (h - 6) / 12)) if 6 <= h <= 18 else 0.0
temps.append(round(t, 1))
ghis.append(round(g, 1))
return temps, ghis
def main():
from src.day_ahead_planner import DayAheadPlanner
from src.data.redis_cache import get_redis
from config.settings import DAILY_PLAN_PATH, MAX_ENERGY_REDUCTION_PCT
from datetime import timedelta
# Use Israel Standard Time (UTC+2) — HF Spaces runs in UTC
IST = timezone(timedelta(hours=2))
target = datetime.now(IST).date()
log.info("Computing day-ahead plan for %s (IST)", target)
# Build forecast inputs
forecast_temps, forecast_ghi = _get_forecast(target)
# Compute a reasonable daily budget (5% of ~25 kWh potential = ~1.25 kWh)
daily_budget_kwh = 25.0 * MAX_ENERGY_REDUCTION_PCT / 100.0
planner = DayAheadPlanner()
plan = planner.plan_day(
target_date=target,
forecast_temps=forecast_temps,
forecast_ghi=forecast_ghi,
daily_budget_kwh=daily_budget_kwh,
)
plan_dict = plan.to_dict() if hasattr(plan, "to_dict") else {"raw": str(plan)}
plan_dict["_computed_at"] = datetime.now(timezone.utc).isoformat()
# Save to file (backup)
try:
Path(DAILY_PLAN_PATH).parent.mkdir(parents=True, exist_ok=True)
with open(DAILY_PLAN_PATH, "w") as f:
json.dump(plan_dict, f, default=str, indent=2)
log.info("Plan saved to %s", DAILY_PLAN_PATH)
except Exception as exc:
log.error("Failed to save plan file: %s", exc)
# Save to Redis
redis = get_redis()
if redis:
safe = json.loads(json.dumps(plan_dict, default=str))
redis.set_json("control:plan", safe, ttl=86400) # 24h TTL
log.info("Plan saved to Redis")
else:
log.warning("Redis not available — plan not shared")
log.info("Plan complete: %d slots", len(plan_dict.get("slots", [])))
if __name__ == "__main__":
main()
|